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

die() {
  echo "error: $*" >&2
  exit 2
}

usage() {
  cat <<'USAGE'
Usage:
  scripts/rerun/run-again.sh --from-attempt PATH [options]
  scripts/rerun/run-again.sh --from-manifest PATH [options]

Creates a new attempt-local run try directory, writes a try-local manifest, and
for an existing predecessor automatically performs mandatory Internet rerun
recovery before it executes the existing public run stage:

  tipoc run --manifest <attempt-root>/run/try-0x/run-profile-manifest.json

Options:
  --from-attempt PATH
  --from-manifest PATH
  --local-attempt-root PATH
  --remote-attempt-root PATH
  --try auto|try-01
  --duration-seconds N
  --warmup-seconds N
  --cooldown-seconds N
  --recovery-timeout-seconds N
  --progress-interval-seconds N
  --concurrency N
  --concurrency-steps N,N,N
  --sql-pool-size N
  --pressure-nodes 'node=addr:capacity:sqlpool[:cecap],...'
  --workload-mix VALUE
  --workload-profile VALUE
  --workload-profile-override FILE
  --exchange-matching-write-percent N
  --exchange-matching-sql-write-percent N
  --query-cost-mode VALUE
  --query-cost-attempt-ru-source VALUE
  --query-cost-attempt-ru-timeout-milliseconds N
  --query-cost-attempt-ru-failure-budget N
  --query-cost-deployment-attestation-path PATH
  --query-cost-deployment-attestation-sha256 SHA256
  --query-cost-poll-interval-seconds N
  --query-cost-poll-timeout-seconds N
  --query-cost-max-leaf-groups N
  --query-cost-max-rollup-groups N
  --query-cost-max-plan-variants N
  --query-cost-isolation-status VALUE
  --query-cost-isolation-reason VALUE
  --ru-acceptance-scenario-id VALUE
  --ru-acceptance-scenario-variant VALUE
  --operation-runtime-overrides FILE
  --tipoc PATH
  --dry-run
  --sync-try-to-pressure-nodes
  --help

The script is an operator wrapper, not a fifth public product stage.
USAGE
}

require_command() {
  command -v "$1" >/dev/null 2>&1 || die "$1 is required"
}

trim_slashes() {
  local value="$1"
  value="${value%/}"
  printf '%s\n' "$value"
}

shell_quote() {
  local value="$1"
  printf "'%s'" "${value//\'/\'\"\'\"\'}"
}

proxy_command_quote() {
  local value="$1"
  if [[ "$value" =~ ^[A-Za-z0-9_./:@%+=,-]+$ ]]; then
    printf '%s' "$value"
    return
  fi
  value="${value//\\/\\\\}"
  value="${value//\"/\\\"}"
  value="${value//\$/\\$}"
  value="${value//\`/\\\`}"
  printf '"%s"' "$value"
}

is_positive_int() {
  [[ "${1:-}" =~ ^[1-9][0-9]*$ ]]
}

is_non_negative_int() {
  [[ "${1:-}" =~ ^[0-9]+$ ]]
}

int_value() {
  printf '%d\n' "$((10#$1))"
}

abs_path() {
  local path="$1"
  if [[ -d "$path" ]]; then
    (cd "$path" && pwd)
  else
    local dir base
    dir="$(dirname "$path")"
    base="$(basename "$path")"
    [[ -d "$dir" ]] || die "path parent directory does not exist: $dir"
    (cd "$dir" && printf '%s/%s\n' "$(pwd)" "$base")
  fi
}

is_try_manifest_path() {
  local path="$1"
  local try_dir run_dir
  try_dir="$(basename "$(dirname "$path")")"
  run_dir="$(basename "$(dirname "$(dirname "$path")")")"
  [[ "$try_dir" =~ ^try-[0-9][0-9]$ && "$run_dir" == "run" ]]
}

remote_attempt_root_from_manifest_path() {
  local path="$1"
  [[ -n "$path" ]] || return 0
  local manifest_dir try_dir run_dir
  manifest_dir="$(dirname "$path")"
  try_dir="$(basename "$manifest_dir")"
  run_dir="$(basename "$(dirname "$manifest_dir")")"
  if [[ "$try_dir" =~ ^try-[0-9][0-9]$ && "$run_dir" == "run" ]]; then
    dirname "$(dirname "$manifest_dir")"
    return
  fi
  printf '%s\n' "$manifest_dir"
}

attempt_root_for_manifest() {
  local manifest="$1"
  local root inferred
  root="$(jq -r '.evidence_root // empty' "$manifest")"
  if [[ -n "$root" && -d "$root" ]]; then
    abs_path "$root"
    return
  fi
  if is_try_manifest_path "$manifest"; then
    inferred="$(dirname "$(dirname "$(dirname "$manifest")")")"
  else
    inferred="$(dirname "$manifest")"
  fi
  if [[ -d "$inferred/prepare" || -d "$inferred" ]]; then
    abs_path "$inferred"
    return
  fi
  if [[ -n "$root" ]]; then
    die "manifest evidence_root is not available locally: $root; pass --local-attempt-root PATH for a local attempt mirror"
  fi
  die "cannot infer local attempt root from manifest path: $manifest"
}

continuation_is_ready() {
  local path="$1"
  [[ -s "$path" ]] || return 1
  jq -e '.status == "ready"' "$path" >/dev/null 2>&1
}

continuation_is_non_ready() {
  local path="$1"
  [[ -s "$path" ]] || return 1
  ! continuation_is_ready "$path"
}

rerun_receipt_is_run_output() {
  local path="$1"
  [[ -s "$path" ]] || return 1
  ! jq -e '.status == "planned"' "$path" >/dev/null 2>&1
}

run_dir_has_output() {
  local dir="$1"
  [[ -s "$dir/run-receipt.json" || -s "$dir/global-run-closeout.json" || -s "$dir/run-report.md" ]] && return 0
  rerun_receipt_is_run_output "$dir/rerun-receipt.json" && return 0
  continuation_is_non_ready "$dir/run-continuation.json" && return 0
  return 1
}

next_try_id() {
  local run_root="$1"
  local max=0 base num
  shopt -s nullglob
  for dir in "$run_root"/try-[0-9][0-9]; do
    base="$(basename "$dir")"
    num="${base#try-}"
    if [[ "$num" =~ ^[0-9]+$ ]] && (( 10#$num > max )); then
      max=$((10#$num))
    fi
  done
  shopt -u nullglob
  printf 'try-%02d\n' "$((max + 1))"
}

latest_completed_try() {
  local run_root="$1"
  local best="" best_num=0 base num
  if continuation_is_ready "$run_root/run-continuation.json" && ( [[ -s "$run_root/run-receipt.json" ]] || [[ -s "$run_root/global-run-closeout.json" ]] ); then
    best="$run_root"
  fi
  shopt -s nullglob
  for dir in "$run_root"/try-[0-9][0-9]; do
    continuation_is_ready "$dir/run-continuation.json" || continue
    [[ -s "$dir/run-receipt.json" || -s "$dir/global-run-closeout.json" ]] || continue
    base="$(basename "$dir")"
    num="${base#try-}"
    if [[ "$num" =~ ^[0-9]+$ ]] && (( 10#$num > best_num )); then
      best_num=$((10#$num))
      best="$dir"
    fi
  done
  shopt -u nullglob
  printf '%s\n' "$best"
}

latest_run_output_try() {
  local run_root="$1"
  local best="" best_num=0 base num
  if run_dir_has_output "$run_root"; then
    best="$run_root"
  fi
  shopt -s nullglob
  for dir in "$run_root"/try-[0-9][0-9]; do
    run_dir_has_output "$dir" || continue
    base="$(basename "$dir")"
    num="${base#try-}"
    if [[ "$num" =~ ^[0-9]+$ ]] && (( 10#$num > best_num )); then
      best_num=$((10#$num))
      best="$dir"
    fi
  done
  shopt -u nullglob
  printf '%s\n' "$best"
}

parse_pressure_nodes_json() {
  local list="$1"
  [[ -n "$list" ]] || die "--pressure-nodes cannot be empty"
  local json='[]'
  local seen=','
  local item node rest address capacity sql_pool ce_cap extra
  IFS=',' read -ra items <<< "$list"
  for item in "${items[@]}"; do
    [[ -n "$item" ]] || continue
    node="${item%%=*}"
    rest="${item#*=}"
    [[ "$node" != "$item" ]] || die "pressure node $item must use node-id=address:capacity:sql-pool[:ce-worker-cap]"
    IFS=':' read -r address capacity sql_pool ce_cap extra <<< "$rest"
    [[ -z "${extra:-}" ]] || die "pressure node $item has too many fields"
    [[ -n "$node" && -n "$address" ]] || die "invalid pressure node $item"
    is_positive_int "$capacity" || die "pressure node $node capacity must be a positive integer"
    is_positive_int "$sql_pool" || die "pressure node $node sql pool must be a positive integer"
    if [[ -n "${ce_cap:-}" ]]; then
      is_positive_int "$ce_cap" || die "pressure node $node CE worker cap must be a positive integer"
    fi
    [[ "$seen" != *",$node,"* ]] || die "duplicate pressure node $node"
    seen="${seen}${node},"
    json="$(jq -c \
      --arg node "$node" \
      --arg address "$address" \
      --argjson capacity "$capacity" \
      --argjson sql_pool "$sql_pool" \
      --arg ce_cap "${ce_cap:-}" \
      '. + [{
        node_id: $node,
        address: $address,
        pressure_capacity_units: $capacity,
        pressure_sql_pool_size: $sql_pool
      } + (if $ce_cap == "" then {} else {pressure_ce_worker_limit: ($ce_cap | tonumber)} end)]' <<< "$json")"
  done
  [[ "$json" != "[]" ]] || die "--pressure-nodes did not contain any nodes"
  printf '%s\n' "$json"
}

pressure_nodes_include_coordinator() {
  local nodes_json="$1"
  local coordinator="$2"
  [[ -n "$nodes_json" ]] || return 0
  [[ -n "$coordinator" ]] || die "--pressure-nodes requires an existing coordinator node in the source manifest"
  jq -e --arg coordinator "$coordinator" '.[] | select(.node_id == $coordinator)' <<< "$nodes_json" >/dev/null 2>&1
}

ssh_args_from_manifest() {
  local manifest="$1"
  local key_ref key_path
  key_ref="$(jq -r '.runtime.runner_launch.ssh_private_key_ref // .scenario.init.ssh_private_key_ref // empty' "$manifest")"
  printf '%s\n' "-o"
  printf '%s\n' "BatchMode=yes"
  printf '%s\n' "-o"
  printf '%s\n' "StrictHostKeyChecking=no"
  printf '%s\n' "-o"
  printf '%s\n' "ConnectTimeout=10"
  if [[ -n "$key_ref" ]]; then
    case "$key_ref" in
      env:*)
        local key_name
        key_name="${key_ref#env:}"
        key_path="${!key_name:-}"
        ;;
      file:*)
        key_path="${key_ref#file:}"
        ;;
      *)
        key_path="$key_ref"
        ;;
    esac
    [[ -n "${key_path:-}" ]] || die "SSH key ref $key_ref resolved to an empty path"
    printf '%s\n' "-i"
    printf '%s\n' "$key_path"
  fi
}

manifest_bastion_from_manifest() {
  local manifest="$1"
  if [[ -n "${TIPOC_RUN_BASTION:-}" ]]; then
    printf '%s\n' "$TIPOC_RUN_BASTION"
    return
  fi
  jq -r '
    .runtime.runner_launch.bastion
    // .runtime.runner_launch.bastion_host
    // .runtime.runner_launch.ssh_bastion
    // .runtime.runner_launch.ssh_jump_host
    // .scenario.init.bastion
    // .scenario.init.bastion_host
    // .scenario.init.ssh_bastion
    // .scenario.init.ssh_jump_host
    // empty
  ' "$manifest"
}

ssh_mode_for_manifest_node() {
  local manifest="$1"
  local node="$2"
  jq -r --arg node "$node" '
    def node_record($list): ([$list[]? | select((.node_id // "") == $node)][0] // {});
    node_record(.runtime.runner_launch.nodes).ssh_mode
    // node_record(.runtime.runner_launch.nodes).sync_ssh_mode
    // node_record(.scenario.pressure.node_specs).ssh_mode
    // node_record(.scenario.pressure.node_specs).sync_ssh_mode
    // empty
  ' "$manifest"
}

is_private_ssh_address() {
  local address="$1"
  case "$address" in
    10.*|192.168.*) return 0 ;;
    172.*)
      local second
      second="${address#172.}"
      second="${second%%.*}"
      [[ "$second" =~ ^[0-9]+$ ]] && (( 10#$second >= 16 && 10#$second <= 31 ))
      return
      ;;
    *) return 1 ;;
  esac
}

node_requires_bastion() {
  local manifest="$1"
  local node="$2"
  local address="$3"
  local bastion="$4"
  local mode
  mode="$(ssh_mode_for_manifest_node "$manifest" "$node")"
  case "$mode" in
    proxy|bastion|jump) return 0 ;;
    direct) return 1 ;;
  esac
  [[ -n "$bastion" ]] && is_private_ssh_address "$address"
}

ssh_args_for_manifest_node() {
  local manifest="$1"
  local node="$2"
  local address="$3"
  local bastion="$4"
  local base_args=()
  local arg proxy_cmd
  while IFS= read -r arg; do
    base_args+=("$arg")
  done < <(ssh_args_from_manifest "$manifest")
  if node_requires_bastion "$manifest" "$node" "$address" "$bastion"; then
    [[ -n "$bastion" ]] || die "pressure node $node requires bastion/proxy SSH but no TIPOC_RUN_BASTION or manifest bastion was configured"
    proxy_cmd="ssh"
    for arg in "${base_args[@]}"; do
      proxy_cmd+=" $(proxy_command_quote "$arg")"
    done
    proxy_cmd+=" -W %h:%p $(proxy_command_quote "$bastion")"
    base_args+=("-o" "ProxyCommand=$proxy_cmd")
  fi
  printf '%s\n' "${base_args[@]}"
}

rsync_ssh_command() {
  local command="ssh"
  local arg
  for arg in "$@"; do
    command+=" $(shell_quote "$arg")"
  done
  printf '%s\n' "$command"
}

manifest_marks_coordinator_try_present() {
  local manifest="$1"
  jq -e '
    (.runtime.runner_launch.coordinator_try_exists == true)
    or (.runtime.runner_launch.coordinator_remote_try_exists == true)
    or (.runtime.runner_launch.coordinator_try_synced == true)
  ' "$manifest" >/dev/null 2>&1
}

coordinator_remote_try_exists() {
  local manifest="$1"
  local remote_try_root="$2"
  local ssh_user="$3"
  local coordinator="$4"
  local bastion="$5"
  local address target
  address="$(jq -r --arg coordinator "$coordinator" '[.scenario.pressure.node_specs[]? | select((.node_id // "") == $coordinator)][0].address // empty' "$manifest")"
  [[ -n "$address" ]] || return 1
  target="${ssh_user}@${address}"
  local coordinator_ssh_args=()
  local arg
  while IFS= read -r arg; do
    coordinator_ssh_args+=("$arg")
  done < <(ssh_args_for_manifest_node "$manifest" "$coordinator" "$address" "$bastion")
  ssh -n "${coordinator_ssh_args[@]}" "$target" "test -d $(shell_quote "$remote_try_root")" >/dev/null 2>&1
}

required_try_sync_files() {
  local try_root="$1"
  local file
  for file in run-profile-manifest.json workload-profile-resolution.json workload-profile-materialization.json rerun-receipt.json; do
    [[ -s "$try_root/$file" ]] || die "local required try file is missing or empty before sync: $try_root/$file"
    printf '%s\n' "$file"
  done
  if [[ -e "$try_root/run-continuation.json" ]]; then
    [[ -s "$try_root/run-continuation.json" ]] || die "local run-continuation.json exists but is empty before sync: $try_root/run-continuation.json"
    printf '%s\n' "run-continuation.json"
  fi
}

sync_try_to_pressure_nodes() {
  local manifest="$1"
  local try_root="$2"
  local explicit_sync="$3"
  local mode remote_try_manifest
  mode="$(jq -r '.runtime.runner_launch.mode // empty' "$manifest")"
  if [[ "$explicit_sync" == "true" ]]; then
    [[ "$mode" == "external" || "$mode" == "coordinator_ssh" ]] || die "--sync-try-to-pressure-nodes requires runtime.runner_launch.mode=external or coordinator_ssh"
  else
    [[ "$mode" == "coordinator_ssh" ]] || return 0
  fi
  remote_try_manifest="$(jq -r '.runtime.runner_launch.remote_manifest_path // empty' "$manifest")"

  local remote_try_root remote_try_parent ssh_user key_ref
  [[ -n "$remote_try_manifest" ]] || die "runtime.runner_launch.remote_manifest_path is required for remote try sync"
  remote_try_root="$(dirname "$remote_try_manifest")"
  remote_try_parent="$(dirname "$remote_try_root")"
  ssh_user="$(jq -r '.runtime.runner_launch.ssh_user // .scenario.init.ssh_user // empty' "$manifest")"
  [[ -n "$ssh_user" ]] || die "runtime.runner_launch.ssh_user or scenario.init.ssh_user is required for remote try sync"
  key_ref="$(jq -r '.runtime.runner_launch.ssh_private_key_ref // .scenario.init.ssh_private_key_ref // empty' "$manifest")"
  if [[ "$explicit_sync" == "true" ]]; then
    [[ -n "$key_ref" ]] || die "runtime.runner_launch.ssh_private_key_ref or scenario.init.ssh_private_key_ref is required for explicit remote try sync"
  fi

  require_command rsync
  require_command ssh

  local nodes_json node_id address target coordinator missing_node_id missing_address
  local bastion skip_coordinator_sync required_files required_file
  bastion="$(manifest_bastion_from_manifest "$manifest")"
  coordinator="$(jq -r '.runtime.runner_launch.coordinator_node_id // .scenario.init.coordinator_node_id // empty' "$manifest")"
  [[ -n "$coordinator" ]] || die "runtime.runner_launch.coordinator_node_id is required for remote try sync"
  skip_coordinator_sync="true"
  if [[ "$explicit_sync" == "true" ]]; then
    skip_coordinator_sync="false"
    if [[ "$(trim_slashes "$remote_try_root")" == "$(trim_slashes "$try_root")" ]]; then
      skip_coordinator_sync="true"
    fi
  fi
  required_files=()
  while IFS= read -r required_file; do
    required_files+=("$required_file")
  done < <(required_try_sync_files "$try_root")
  missing_node_id="$(jq -r '.scenario.pressure.node_specs[]? | select((.node_id // "") == "") | "missing"' "$manifest" | head -n 1)"
  [[ -z "$missing_node_id" ]] || die "scenario.pressure.node_specs[].node_id is required for remote try sync"
  if [[ "$skip_coordinator_sync" == "true" ]]; then
    missing_address="$(jq -r --arg coordinator "$coordinator" '.scenario.pressure.node_specs[]? | select((.node_id // "") != $coordinator) | select((.address // "") == "") | .node_id' "$manifest" | head -n 1)"
  else
    jq -e --arg coordinator "$coordinator" '.scenario.pressure.node_specs[]? | select((.node_id // "") == $coordinator)' "$manifest" >/dev/null || die "scenario.pressure.node_specs must include coordinator node $coordinator for explicit local-to-remote sync"
    missing_address="$(jq -r '.scenario.pressure.node_specs[]? | select((.address // "") == "") | .node_id' "$manifest" | head -n 1)"
  fi
  [[ -z "$missing_address" ]] || die "scenario.pressure.node_specs address is required for pressure node $missing_address"
  nodes_json="$(jq -c --arg coordinator "$coordinator" --arg skip_coordinator_sync "$skip_coordinator_sync" '[.scenario.pressure.node_specs[]? | select(($skip_coordinator_sync != "true") or ((.node_id // "") != $coordinator)) | {node_id, address}]' "$manifest")"
  [[ "$nodes_json" != "[]" ]] || die "remote try sync requires at least one target pressure node"

  while IFS=$'\t' read -r node_id address; do
    [[ -n "$node_id" && -n "$address" ]] || continue
    local node_ssh_args=()
    local rsync_ssh arg
    while IFS= read -r arg; do
      node_ssh_args+=("$arg")
    done < <(ssh_args_for_manifest_node "$manifest" "$node_id" "$address" "$bastion")
    rsync_ssh="$(rsync_ssh_command "${node_ssh_args[@]}")"
    target="${ssh_user}@${address}"
    echo "sync_node=${node_id} status=syncing"
    if ! ssh -n "${node_ssh_args[@]}" "$target" "mkdir -p $(shell_quote "$remote_try_parent") && mkdir $(shell_quote "$remote_try_root")"; then
      die "sync_node=${node_id} could not exclusively claim absent remote try directory: $remote_try_root"
    fi
    rsync -az -e "$rsync_ssh" "$try_root/" "$target:$remote_try_root/"
    for required_file in "${required_files[@]}"; do
      if ! ssh -n "${node_ssh_args[@]}" "$target" "test -s $(shell_quote "$remote_try_root/$required_file")"; then
        die "sync_node=${node_id} missing required remote try file after sync: $required_file"
      fi
    done
    echo "sync_node=${node_id} status=ok"
  done < <(jq -r '.[] | [.node_id, .address] | @tsv' <<< "$nodes_json")
}

from_attempt=""
from_manifest=""
local_attempt_root=""
remote_attempt_root_option=""
try_id="auto"
duration_seconds=""
warmup_seconds=""
cooldown_seconds=""
recovery_timeout_seconds=""
progress_interval_seconds=""
concurrency=""
concurrency_steps=""
sql_pool_size=""
pressure_nodes=""
workload_mix=""
workload_profile=""
workload_profile_override=""
exchange_matching_write_percent=""
exchange_matching_sql_write_percent=""
query_cost_mode=""
query_cost_attempt_ru_source=""
query_cost_attempt_ru_timeout_milliseconds=""
query_cost_attempt_ru_failure_budget=""
query_cost_deployment_attestation_path=""
query_cost_deployment_attestation_sha256=""
query_cost_poll_interval_seconds=""
query_cost_poll_timeout_seconds=""
query_cost_max_leaf_groups=""
query_cost_max_rollup_groups=""
query_cost_max_plan_variants=""
query_cost_isolation_status=""
query_cost_isolation_reason=""
ru_acceptance_scenario_id=""
ru_acceptance_scenario_variant=""
operation_runtime_overrides=""
tipoc_bin="tipoc"
dry_run="false"
sync_try_to_pressure_nodes_flag="false"

while [[ $# -gt 0 ]]; do
  case "$1" in
    --from-attempt) from_attempt="$2"; shift 2 ;;
    --from-manifest) from_manifest="$2"; shift 2 ;;
    --local-attempt-root) local_attempt_root="$2"; shift 2 ;;
    --remote-attempt-root) remote_attempt_root_option="$2"; shift 2 ;;
    --try) try_id="$2"; shift 2 ;;
    --duration-seconds) duration_seconds="$2"; shift 2 ;;
    --warmup-seconds) warmup_seconds="$2"; shift 2 ;;
    --cooldown-seconds) cooldown_seconds="$2"; shift 2 ;;
    --recovery-timeout-seconds) recovery_timeout_seconds="$2"; shift 2 ;;
    --progress-interval-seconds) progress_interval_seconds="$2"; shift 2 ;;
    --concurrency) concurrency="$2"; shift 2 ;;
    --concurrency-steps) concurrency_steps="$2"; shift 2 ;;
    --sql-pool-size) sql_pool_size="$2"; shift 2 ;;
    --pressure-nodes) pressure_nodes="$2"; shift 2 ;;
    --workload-mix) workload_mix="$2"; shift 2 ;;
    --workload-profile) workload_profile="$2"; shift 2 ;;
    --workload-profile-override) workload_profile_override="$2"; shift 2 ;;
    --exchange-matching-write-percent) exchange_matching_write_percent="$2"; shift 2 ;;
    --exchange-matching-sql-write-percent) exchange_matching_sql_write_percent="$2"; shift 2 ;;
    --query-cost-mode) query_cost_mode="$2"; shift 2 ;;
    --query-cost-attempt-ru-source) query_cost_attempt_ru_source="$2"; shift 2 ;;
    --query-cost-attempt-ru-timeout-milliseconds) query_cost_attempt_ru_timeout_milliseconds="$2"; shift 2 ;;
    --query-cost-attempt-ru-failure-budget) query_cost_attempt_ru_failure_budget="$2"; shift 2 ;;
    --query-cost-deployment-attestation-path) query_cost_deployment_attestation_path="$2"; shift 2 ;;
    --query-cost-deployment-attestation-sha256) query_cost_deployment_attestation_sha256="$2"; shift 2 ;;
    --query-cost-poll-interval-seconds) query_cost_poll_interval_seconds="$2"; shift 2 ;;
    --query-cost-poll-timeout-seconds) query_cost_poll_timeout_seconds="$2"; shift 2 ;;
    --query-cost-max-leaf-groups) query_cost_max_leaf_groups="$2"; shift 2 ;;
    --query-cost-max-rollup-groups) query_cost_max_rollup_groups="$2"; shift 2 ;;
    --query-cost-max-plan-variants) query_cost_max_plan_variants="$2"; shift 2 ;;
    --query-cost-isolation-status) query_cost_isolation_status="$2"; shift 2 ;;
    --query-cost-isolation-reason) query_cost_isolation_reason="$2"; shift 2 ;;
    --ru-acceptance-scenario-id) ru_acceptance_scenario_id="$2"; shift 2 ;;
    --ru-acceptance-scenario-variant) ru_acceptance_scenario_variant="$2"; shift 2 ;;
    --operation-runtime-overrides) operation_runtime_overrides="$2"; shift 2 ;;
    --tipoc) tipoc_bin="$2"; shift 2 ;;
    --dry-run) dry_run="true"; shift ;;
    --sync-try-to-pressure-nodes) sync_try_to_pressure_nodes_flag="true"; shift ;;
    --help|-h) usage; exit 0 ;;
    *) die "unknown option $1" ;;
  esac
done

if [[ -n "$query_cost_deployment_attestation_path" || -n "$query_cost_deployment_attestation_sha256" ]]; then
  [[ -n "$query_cost_deployment_attestation_path" && -n "$query_cost_deployment_attestation_sha256" ]] || die "--query-cost-deployment-attestation-path and --query-cost-deployment-attestation-sha256 must be provided together"
fi

require_command jq

[[ -n "$from_attempt" || -n "$from_manifest" ]] || die "one of --from-attempt or --from-manifest is required"
[[ -z "$from_attempt" || -z "$from_manifest" ]] || die "use only one of --from-attempt or --from-manifest"

if [[ -n "$from_attempt" ]]; then
  source_attempt_root="$(abs_path "$from_attempt")"
  source_manifest="$source_attempt_root/run-profile-manifest.json"
else
  source_manifest="$(abs_path "$from_manifest")"
fi
[[ -s "$source_manifest" ]] || die "source manifest not found: $source_manifest"
if [[ -n "$workload_profile_override" ]]; then
  [[ -s "$workload_profile_override" ]] || die "workload profile override not found: $workload_profile_override"
  workload_profile_override="$(abs_path "$workload_profile_override")"
fi
if [[ -n "$operation_runtime_overrides" ]]; then
  [[ -s "$operation_runtime_overrides" ]] || die "operation runtime override not found: $operation_runtime_overrides"
  operation_runtime_overrides="$(abs_path "$operation_runtime_overrides")"
fi
if [[ -n "$workload_profile" && -n "$workload_profile_override" ]]; then
  die "use either --workload-profile or --workload-profile-override, not both"
fi
if [[ -n "$ru_acceptance_scenario_id" || -n "$ru_acceptance_scenario_variant" ]]; then
  [[ -n "$ru_acceptance_scenario_id" && -n "$ru_acceptance_scenario_variant" ]] || die "--ru-acceptance-scenario-id and --ru-acceptance-scenario-variant must be provided together"
fi
if [[ -n "$local_attempt_root" ]]; then
  attempt_root="$(abs_path "$local_attempt_root")"
  if [[ -n "${source_attempt_root:-}" && "$(trim_slashes "$attempt_root")" != "$(trim_slashes "$source_attempt_root")" ]]; then
    die "--local-attempt-root must match --from-attempt when both are provided"
  fi
elif [[ -n "${source_attempt_root:-}" ]]; then
  attempt_root="$source_attempt_root"
else
  attempt_root="$(attempt_root_for_manifest "$source_manifest")"
fi
[[ -d "$attempt_root/prepare" ]] || die "prepare evidence directory not found under local attempt root: $attempt_root/prepare"

for value in \
  ${duration_seconds:+"duration:$duration_seconds:positive"} \
  ${warmup_seconds:+"warmup:$warmup_seconds:nonnegative"} \
  ${cooldown_seconds:+"cooldown:$cooldown_seconds:nonnegative"} \
  ${progress_interval_seconds:+"progress:$progress_interval_seconds:positive"} \
  ${concurrency:+"concurrency:$concurrency:positive"} \
  ${sql_pool_size:+"sql-pool-size:$sql_pool_size:positive"} \
  ${query_cost_attempt_ru_timeout_milliseconds:+"query-cost-attempt-ru-timeout-milliseconds:$query_cost_attempt_ru_timeout_milliseconds:positive"} \
  ${query_cost_attempt_ru_failure_budget:+"query-cost-attempt-ru-failure-budget:$query_cost_attempt_ru_failure_budget:positive"} \
  ${query_cost_poll_interval_seconds:+"query-cost-poll-interval-seconds:$query_cost_poll_interval_seconds:positive"} \
  ${query_cost_poll_timeout_seconds:+"query-cost-poll-timeout-seconds:$query_cost_poll_timeout_seconds:positive"} \
  ${query_cost_max_leaf_groups:+"query-cost-max-leaf-groups:$query_cost_max_leaf_groups:positive"} \
  ${query_cost_max_rollup_groups:+"query-cost-max-rollup-groups:$query_cost_max_rollup_groups:positive"} \
  ${query_cost_max_plan_variants:+"query-cost-max-plan-variants:$query_cost_max_plan_variants:positive"}; do
  name="${value%%:*}"
  rest="${value#*:}"
  number="${rest%%:*}"
  kind="${rest##*:}"
  if [[ "$kind" == "positive" ]]; then
    is_positive_int "$number" || die "--$name must be a positive integer"
  else
    is_non_negative_int "$number" || die "--$name must be a non-negative integer"
  fi
done

if [[ -n "$query_cost_attempt_ru_timeout_milliseconds" ]]; then
  query_cost_attempt_ru_timeout_value="$(int_value "$query_cost_attempt_ru_timeout_milliseconds")"
  (( query_cost_attempt_ru_timeout_value <= 30000 )) || die "--query-cost-attempt-ru-timeout-milliseconds must be at most 30000"
fi
if [[ -n "$query_cost_attempt_ru_failure_budget" ]]; then
  query_cost_attempt_ru_failure_budget_value="$(int_value "$query_cost_attempt_ru_failure_budget")"
  (( query_cost_attempt_ru_failure_budget_value <= 1000000 )) || die "--query-cost-attempt-ru-failure-budget must be at most 1000000"
fi
if [[ -n "$query_cost_max_leaf_groups" ]]; then
  query_cost_max_leaf_groups_value="$(int_value "$query_cost_max_leaf_groups")"
  (( query_cost_max_leaf_groups_value <= 4096 )) || die "--query-cost-max-leaf-groups must be at most 4096"
fi
if [[ -n "$query_cost_max_rollup_groups" ]]; then
  query_cost_max_rollup_groups_value="$(int_value "$query_cost_max_rollup_groups")"
  (( query_cost_max_rollup_groups_value <= 8192 )) || die "--query-cost-max-rollup-groups must be at most 8192"
fi
if [[ -n "$query_cost_max_leaf_groups" || -n "$query_cost_max_rollup_groups" ]]; then
  source_query_cost_max_leaf_groups="$(jq -r '.scenario.reporting.query_cost.max_leaf_groups // empty' "$source_manifest")"
  source_query_cost_contract_max_leaf_groups="$(jq -r '.fullflow_profile_contract.reporting.query_cost.max_leaf_groups // empty' "$source_manifest")"
  source_query_cost_max_rollup_groups="$(jq -r '.scenario.reporting.query_cost.max_rollup_groups // empty' "$source_manifest")"
  source_query_cost_contract_max_rollup_groups="$(jq -r '.fullflow_profile_contract.reporting.query_cost.max_rollup_groups // empty' "$source_manifest")"
  [[ "$source_query_cost_max_leaf_groups" == "$source_query_cost_contract_max_leaf_groups" ]] || die "source query-cost scenario and frozen contract leaf capacities differ"
  [[ "$source_query_cost_max_rollup_groups" == "$source_query_cost_contract_max_rollup_groups" ]] || die "source query-cost scenario and frozen contract rollup capacities differ"
  final_query_cost_max_leaf_groups="${query_cost_max_leaf_groups:-$source_query_cost_max_leaf_groups}"
  final_query_cost_max_rollup_groups="${query_cost_max_rollup_groups:-$source_query_cost_max_rollup_groups}"
  is_positive_int "$final_query_cost_max_leaf_groups" || die "final query-cost leaf capacity must be a positive integer"
  is_positive_int "$final_query_cost_max_rollup_groups" || die "final query-cost rollup capacity must be a positive integer"
  final_query_cost_max_leaf_groups_value="$(int_value "$final_query_cost_max_leaf_groups")"
  final_query_cost_max_rollup_groups_value="$(int_value "$final_query_cost_max_rollup_groups")"
  (( final_query_cost_max_rollup_groups_value >= final_query_cost_max_leaf_groups_value )) || die "query-cost rollup capacity must be at least leaf capacity"
fi

if [[ -n "$query_cost_poll_interval_seconds" || -n "$query_cost_poll_timeout_seconds" || -n "$query_cost_max_plan_variants" ]]; then
  source_query_cost_mode="$(jq -r '.scenario.reporting.query_cost.mode // "off"' "$source_manifest")"
  source_query_cost_contract_mode="$(jq -r '.fullflow_profile_contract.reporting.query_cost.mode // "off"' "$source_manifest")"
  [[ "$source_query_cost_mode" != "off" ]] || die "query-cost overrides require query cost to be enabled in the source manifest"
  [[ "$source_query_cost_contract_mode" == "$source_query_cost_mode" ]] || die "source query-cost scenario and frozen contract modes differ"
  source_query_cost_poll_interval="$(jq -r '.scenario.reporting.query_cost.poll_interval_seconds // empty' "$source_manifest")"
  source_query_cost_contract_poll_interval="$(jq -r '.fullflow_profile_contract.reporting.query_cost.poll_interval_seconds // empty' "$source_manifest")"
  source_query_cost_poll_timeout="$(jq -r '.scenario.reporting.query_cost.poll_timeout_seconds // empty' "$source_manifest")"
  source_query_cost_contract_poll_timeout="$(jq -r '.fullflow_profile_contract.reporting.query_cost.poll_timeout_seconds // empty' "$source_manifest")"
  source_query_cost_max_plan_variants="$(jq -r '.scenario.reporting.query_cost.max_plan_variants // empty' "$source_manifest")"
  source_query_cost_contract_max_plan_variants="$(jq -r '.fullflow_profile_contract.reporting.query_cost.max_plan_variants // empty' "$source_manifest")"
  [[ "$source_query_cost_poll_interval" == "$source_query_cost_contract_poll_interval" ]] || die "source query-cost scenario and frozen contract poll intervals differ"
  [[ "$source_query_cost_poll_timeout" == "$source_query_cost_contract_poll_timeout" ]] || die "source query-cost scenario and frozen contract poll timeouts differ"
  [[ "$source_query_cost_max_plan_variants" == "$source_query_cost_contract_max_plan_variants" ]] || die "source query-cost scenario and frozen contract plan-variant limits differ"
  final_query_cost_poll_interval="${query_cost_poll_interval_seconds:-$source_query_cost_poll_interval}"
  final_query_cost_poll_timeout="${query_cost_poll_timeout_seconds:-$source_query_cost_poll_timeout}"
  is_positive_int "$final_query_cost_poll_interval" || die "final query-cost poll interval must be a positive integer"
  is_positive_int "$final_query_cost_poll_timeout" || die "final query-cost poll timeout must be a positive integer"
  final_query_cost_poll_interval_value="$(int_value "$final_query_cost_poll_interval")"
  final_query_cost_poll_timeout_value="$(int_value "$final_query_cost_poll_timeout")"
  (( final_query_cost_poll_interval_value >= 5 )) || die "--query-cost-poll-interval-seconds must resolve to at least 5"
  (( final_query_cost_poll_timeout_value < final_query_cost_poll_interval_value )) || die "--query-cost-poll-timeout-seconds must resolve below the poll interval"
  if [[ -n "$query_cost_max_plan_variants" ]]; then
    query_cost_max_plan_variants_value="$(int_value "$query_cost_max_plan_variants")"
    (( query_cost_max_plan_variants_value <= 256 )) || die "--query-cost-max-plan-variants must be at most 256"
  fi
fi

if [[ -n "$query_cost_isolation_status" || -n "$query_cost_isolation_reason" ]]; then
  [[ -n "$query_cost_isolation_status" && -n "$query_cost_isolation_reason" ]] || die "query-cost isolation status and reason must be overridden together"
  case "$query_cost_isolation_status" in
    verified_test_environment|declared|unverified) ;;
    *) die "--query-cost-isolation-status must be verified_test_environment, declared, or unverified" ;;
  esac
  [[ "${#query_cost_isolation_reason}" -le 1024 ]] || die "--query-cost-isolation-reason must not exceed 1024 bytes"
  source_query_cost_mode="$(jq -r '.scenario.reporting.query_cost.mode // "off"' "$source_manifest")"
  source_query_cost_contract_mode="$(jq -r '.fullflow_profile_contract.reporting.query_cost.mode // "off"' "$source_manifest")"
  [[ "$source_query_cost_mode" != "off" ]] || die "query-cost isolation override requires query cost to be enabled in the source manifest"
  [[ "$source_query_cost_contract_mode" == "$source_query_cost_mode" ]] || die "source query-cost scenario and frozen contract modes differ"
fi

if [[ -n "$exchange_matching_write_percent" ]]; then
  is_non_negative_int "$exchange_matching_write_percent" || die "--exchange-matching-write-percent must be an integer between 20 and 90"
  exchange_matching_write_percent_value="$(int_value "$exchange_matching_write_percent")"
  (( exchange_matching_write_percent_value >= 20 && exchange_matching_write_percent_value <= 90 )) || die "--exchange-matching-write-percent must be between 20 and 90"
fi
if [[ -n "$exchange_matching_sql_write_percent" ]]; then
  is_non_negative_int "$exchange_matching_sql_write_percent" || die "--exchange-matching-sql-write-percent must be 0 or an integer between 10 and 90"
  exchange_matching_sql_write_percent_value="$(int_value "$exchange_matching_sql_write_percent")"
  (( exchange_matching_sql_write_percent_value == 0 || (exchange_matching_sql_write_percent_value >= 10 && exchange_matching_sql_write_percent_value <= 90) )) || die "--exchange-matching-sql-write-percent must be 0 or between 10 and 90"
fi
if [[ -n "$exchange_matching_write_percent" && -z "$exchange_matching_sql_write_percent" ]]; then
  die "--exchange-matching-write-percent changes only the CE selector ratio; it is not the observed SQL write target. Pass --exchange-matching-sql-write-percent N, or pass --exchange-matching-sql-write-percent 0 only when intentionally disabling SQL write target shaping."
fi

if [[ -n "$exchange_matching_write_percent" || -n "$exchange_matching_sql_write_percent" ]]; then
  if [[ -n "$workload_mix" ]]; then
    final_runtime_workload_mix="$workload_mix"
    final_pressure_workload_mix="$workload_mix"
  else
    final_runtime_workload_mix="$(jq -r '.runtime.workload_mix // empty' "$source_manifest")"
    final_pressure_workload_mix="$(jq -r '.scenario.pressure.workload_mix // empty' "$source_manifest")"
  fi
  if [[ "$final_runtime_workload_mix" != "exchange_matching" || "$final_pressure_workload_mix" != "exchange_matching" ]]; then
    die "--exchange-matching-write-percent and --exchange-matching-sql-write-percent require final runtime.workload_mix and scenario.pressure.workload_mix to be exchange_matching"
  fi
fi

step_values=()
if [[ -n "$concurrency_steps" ]]; then
  [[ -z "$concurrency" ]] || die "use either --concurrency or --concurrency-steps, not both"
  IFS=',' read -ra step_values <<< "$concurrency_steps"
  [[ "${#step_values[@]}" -gt 0 ]] || die "--concurrency-steps cannot be empty"
  for step in "${step_values[@]}"; do
    is_positive_int "$step" || die "each --concurrency-steps value must be a positive integer"
  done
else
  step_values=("${concurrency:-}")
fi

pressure_nodes_json=""
if [[ -n "$pressure_nodes" ]]; then
  pressure_nodes_json="$(parse_pressure_nodes_json "$pressure_nodes")"
  source_coordinator_node_id="$(jq -r '.runtime.runner_launch.coordinator_node_id // .scenario.init.coordinator_node_id // empty' "$source_manifest")"
  if ! pressure_nodes_include_coordinator "$pressure_nodes_json" "$source_coordinator_node_id"; then
    die "--pressure-nodes must include the current coordinator node $source_coordinator_node_id; create a new attempt if the coordinator changes"
  fi
fi

run_root="$attempt_root/run"
mkdir -p "$run_root"

if [[ -n "$remote_attempt_root_option" ]]; then
  remote_attempt_root="$remote_attempt_root_option"
else
  remote_attempt_root="$(jq -r '.scenario.init.remote_attempt_root // .init.remote_attempt_root // empty' "$source_manifest")"
  remote_attempt_root="$(trim_slashes "$remote_attempt_root")"
  if [[ -z "$remote_attempt_root" ]]; then
    remote_source_manifest="$(jq -r '.runtime.runner_launch.remote_manifest_path // empty' "$source_manifest")"
    remote_attempt_root="$(remote_attempt_root_from_manifest_path "$remote_source_manifest")"
    remote_attempt_root="$(trim_slashes "$remote_attempt_root")"
  fi
  if [[ -z "$remote_attempt_root" ]]; then
    manifest_evidence_root="$(jq -r '.evidence_root // empty' "$source_manifest")"
    manifest_evidence_root="$(trim_slashes "$manifest_evidence_root")"
    if [[ -n "$manifest_evidence_root" && "$manifest_evidence_root" != "$(trim_slashes "$attempt_root")" ]]; then
      remote_attempt_root="$manifest_evidence_root"
    fi
  fi
fi
remote_attempt_root="$(trim_slashes "$remote_attempt_root")"
if [[ "$sync_try_to_pressure_nodes_flag" == "true" && -z "$remote_attempt_root" ]]; then
  die "--sync-try-to-pressure-nodes requires --remote-attempt-root or a manifest remote attempt root"
fi

for idx in "${!step_values[@]}"; do
  step_concurrency="${step_values[$idx]}"
  selected_try="$try_id"
  if [[ "$selected_try" == "auto" ]]; then
    selected_try="$(next_try_id "$run_root")"
  elif [[ "${#step_values[@]}" -gt 1 ]]; then
    die "explicit --try cannot be combined with --concurrency-steps"
  fi
  [[ "$selected_try" =~ ^try-[0-9][0-9]$ ]] || die "--try must be auto or try-01 style"

  try_root="$run_root/$selected_try"
  if [[ -e "$try_root" || -L "$try_root" ]]; then
    die "$try_root already exists; a new run try requires a previously absent directory"
  fi

  previous_try="$(latest_completed_try "$run_root")"
  latest_output_try="$(latest_run_output_try "$run_root")"
	# The internal `run recover` capability is implemented only for the
	# Internet workload.  Other workload families retain their established Run
	# continuation path and must never be routed through Internet recovery.
	pre_run_recovery_required="false"
  # A missing terminal boundary can be caused by an ungraceful process stop.
  # Still materialize the next isolated try and let the public run stage derive
  # a safe cursor from sealed evidence and live runtime state. This wrapper has
  # no authority to classify or repair a non-ready predecessor itself.
  if [[ -n "$previous_try" && -s "$previous_try/run-continuation.json" ]]; then
    continuation_source="$previous_try/run-continuation.json"
  else
    continuation_source=""
    source_state="$(jq -r '.stage.current_state // empty' "$source_manifest")"
    # The public run stage owns recovery for a missing boundary.  Do not reject
    # here: this wrapper has no authority to inspect target runtime state.
  fi

  target_manifest="$try_root/run-profile-manifest.json"
  remote_try_manifest=""
  remote_try_log_dir=""
  manifest_attempt_root="$attempt_root"
  manifest_try_root="$try_root"
  remote_manifest_paths_enabled="false"
  if [[ -n "$remote_attempt_root" ]]; then
    remote_try_manifest="$remote_attempt_root/run/$selected_try/run-profile-manifest.json"
    remote_try_log_dir="$remote_attempt_root/logs/run/$selected_try"
    if [[ "$sync_try_to_pressure_nodes_flag" == "true" ]]; then
      manifest_attempt_root="$remote_attempt_root"
      manifest_try_root="$remote_attempt_root/run/$selected_try"
      remote_manifest_paths_enabled="true"
    fi
  fi
  concurrency_value="$step_concurrency"
  materializer_args=(
    internal workload-profile materialize
    --manifest "$target_manifest"
    --manifest-template "$source_manifest"
    --attempt-root "$manifest_attempt_root"
    --try-root "$manifest_try_root"
    --continuation-source "$continuation_source"
    --remote-try-manifest "$remote_try_manifest"
    --remote-try-log-dir "$remote_try_log_dir"
    --duration-seconds "$duration_seconds"
    --warmup-seconds "$warmup_seconds"
    --cooldown-seconds "$cooldown_seconds"
    --recovery-timeout-seconds "$recovery_timeout_seconds"
    --progress-interval-seconds "$progress_interval_seconds"
    --query-cost-mode "$query_cost_mode"
    --query-cost-attempt-ru-source "$query_cost_attempt_ru_source"
    --query-cost-attempt-ru-timeout-milliseconds "$query_cost_attempt_ru_timeout_milliseconds"
    --query-cost-attempt-ru-failure-budget "$query_cost_attempt_ru_failure_budget"
    --query-cost-deployment-attestation-path "$query_cost_deployment_attestation_path"
    --query-cost-deployment-attestation-sha256 "$query_cost_deployment_attestation_sha256"
    --query-cost-poll-interval-seconds "$query_cost_poll_interval_seconds"
    --query-cost-poll-timeout-seconds "$query_cost_poll_timeout_seconds"
    --query-cost-max-leaf-groups "$query_cost_max_leaf_groups"
    --query-cost-max-rollup-groups "$query_cost_max_rollup_groups"
    --query-cost-max-plan-variants "$query_cost_max_plan_variants"
    --query-cost-isolation-status "$query_cost_isolation_status"
    --query-cost-isolation-reason "$query_cost_isolation_reason"
    --ru-acceptance-scenario-id "$ru_acceptance_scenario_id"
    --ru-acceptance-scenario-variant "$ru_acceptance_scenario_variant"
    --concurrency "$concurrency_value"
    --workload-mix "$workload_mix"
    --exchange-matching-write-percent "$exchange_matching_write_percent"
    --exchange-matching-sql-write-percent "$exchange_matching_sql_write_percent"
    --pressure-nodes-json "$pressure_nodes_json"
    --sql-pool-size "$sql_pool_size"
  )
  if [[ -n "$workload_profile_override" ]]; then
    materializer_args+=(--workload-profile-override "$workload_profile_override")
  fi
  if [[ -n "$workload_profile" ]]; then
    materializer_args+=(--workload-profile "$workload_profile")
  fi
  "$tipoc_bin" "${materializer_args[@]}"
  workload_profile_resolution="$try_root/workload-profile-resolution.json"
  workload_profile_materialization="$try_root/workload-profile-materialization.json"
  [[ -s "$workload_profile_resolution" ]] || die "workload profile materializer did not write resolution evidence: $workload_profile_resolution"
  [[ -s "$workload_profile_materialization" ]] || die "workload profile materializer did not write commit evidence: $workload_profile_materialization"
	if [[ -n "$latest_output_try" ]] && [[ "$(jq -r '.workload.industry // empty' "$target_manifest")" == "internet" ]]; then
		pre_run_recovery_required="true"
	fi
  if [[ -n "$continuation_source" ]]; then
    cp "$continuation_source" "$try_root/run-continuation.json"
  fi

  receipt="$try_root/rerun-receipt.json"
  jq -n \
    --arg schema_version "tipoc.v3.rerun-receipt.v1" \
    --arg attempt_root "$attempt_root" \
    --arg remote_attempt_root "$remote_attempt_root" \
    --arg source_manifest "$source_manifest" \
    --arg target_manifest "$target_manifest" \
    --arg remote_target_manifest "$remote_try_manifest" \
    --arg try_id "$selected_try" \
    --arg continuation_source "$continuation_source" \
		--arg pre_run_recovery_required "$pre_run_recovery_required" \
    --arg dry_run "$dry_run" \
    --arg workload_profile_override "$workload_profile_override" \
    --arg workload_profile_materialization "$workload_profile_materialization" \
    --argjson workload_profile_resolution "$(jq -c '.' "$workload_profile_resolution")" \
    --argjson overrides "$(jq -n \
      --arg duration "$duration_seconds" \
      --arg warmup "$warmup_seconds" \
      --arg cooldown "$cooldown_seconds" \
      --arg progress "$progress_interval_seconds" \
      --arg concurrency "$concurrency_value" \
      --arg sql_pool_size "$sql_pool_size" \
      --arg pressure_nodes "$pressure_nodes" \
      --arg workload_mix "$workload_mix" \
      --arg workload_profile "$workload_profile" \
      --arg workload_profile_override "$workload_profile_override" \
      --arg exchange_matching_write_percent "$exchange_matching_write_percent" \
      --arg exchange_matching_sql_write_percent "$exchange_matching_sql_write_percent" \
      --arg query_cost_mode "$query_cost_mode" \
      --arg query_cost_attempt_ru_source "$query_cost_attempt_ru_source" \
      --arg query_cost_attempt_ru_timeout_milliseconds "$query_cost_attempt_ru_timeout_milliseconds" \
      --arg query_cost_attempt_ru_failure_budget "$query_cost_attempt_ru_failure_budget" \
      --arg query_cost_deployment_attestation_sha256 "$query_cost_deployment_attestation_sha256" \
      --arg query_cost_poll_interval_seconds "$query_cost_poll_interval_seconds" \
      --arg query_cost_poll_timeout_seconds "$query_cost_poll_timeout_seconds" \
      --arg query_cost_max_leaf_groups "$query_cost_max_leaf_groups" \
      --arg query_cost_max_rollup_groups "$query_cost_max_rollup_groups" \
      --arg query_cost_max_plan_variants "$query_cost_max_plan_variants" \
      --arg query_cost_isolation_status "$query_cost_isolation_status" \
      --arg query_cost_isolation_reason "$query_cost_isolation_reason" \
      --arg ru_acceptance_scenario_id "$ru_acceptance_scenario_id" \
      --arg ru_acceptance_scenario_variant "$ru_acceptance_scenario_variant" \
      --arg operation_runtime_overrides "$operation_runtime_overrides" \
      '{
        duration_seconds: $duration,
        warmup_seconds: $warmup,
        cooldown_seconds: $cooldown,
        progress_interval_seconds: $progress,
        concurrency: $concurrency,
        sql_pool_size: $sql_pool_size,
        pressure_nodes: $pressure_nodes,
        workload_mix: $workload_mix,
        workload_profile: $workload_profile,
        workload_profile_override: $workload_profile_override,
        exchange_matching_write_percent: $exchange_matching_write_percent,
        exchange_matching_sql_write_percent: $exchange_matching_sql_write_percent,
        query_cost_mode: $query_cost_mode,
        query_cost_attempt_ru_source: $query_cost_attempt_ru_source,
        query_cost_attempt_ru_timeout_milliseconds: $query_cost_attempt_ru_timeout_milliseconds,
        query_cost_attempt_ru_failure_budget: $query_cost_attempt_ru_failure_budget,
        query_cost_deployment_attestation_sha256: $query_cost_deployment_attestation_sha256,
        query_cost_poll_interval_seconds: $query_cost_poll_interval_seconds,
        query_cost_poll_timeout_seconds: $query_cost_poll_timeout_seconds,
        query_cost_max_leaf_groups: $query_cost_max_leaf_groups,
        query_cost_max_rollup_groups: $query_cost_max_rollup_groups,
        query_cost_max_plan_variants: $query_cost_max_plan_variants,
        query_cost_isolation_status: $query_cost_isolation_status,
        query_cost_isolation_reason: $query_cost_isolation_reason,
        ru_acceptance_scenario_id: $ru_acceptance_scenario_id,
        ru_acceptance_scenario_variant: $ru_acceptance_scenario_variant,
        operation_runtime_overrides: $operation_runtime_overrides
      } | with_entries(select(.value != ""))')" \
    '{
      schema_version: $schema_version,
      status: (if $dry_run == "true" then "planned" else "running" end),
      attempt_root: $attempt_root,
      local_attempt_root: $attempt_root,
      remote_attempt_root: $remote_attempt_root,
      try_id: $try_id,
      source_manifest: $source_manifest,
      target_manifest: $target_manifest,
      remote_target_manifest: $remote_target_manifest,
      continuation_source: $continuation_source,
      continuation_mode: (if $continuation_source == "" then "from_prepare_boundary" else "from_previous_try_boundary" end),
		pre_run_recovery: { required: ($pre_run_recovery_required == "true"), status: "not_started" },
      overrides: $overrides,
      workload_profile_override: $workload_profile_override,
      workload_profile_materialization: $workload_profile_materialization,
      workload_profile_resolution: $workload_profile_resolution
    }' > "$receipt"

  echo "try=${selected_try}"
  echo "manifest=${target_manifest}"
  echo "run_manifest=${target_manifest}"
  echo "receipt=${receipt}"
  echo "workload_profile_resolution=${workload_profile_resolution}"
  echo "workload_profile_materialization=${workload_profile_materialization}"
  if [[ "$remote_manifest_paths_enabled" == "true" && "$dry_run" != "true" ]]; then
    jq '.status = "blocked" | .blocker = "remote-compatible manifest paths require remote launch; local non-dry run is blocked to avoid writing remote-only output paths from this host" | .exit_code = 2' "$receipt" > "${receipt}.tmp" && mv "${receipt}.tmp" "$receipt"
    die "--sync-try-to-pressure-nodes generated remote-compatible manifest paths; use --dry-run for remote handoff or launch the remote manifest on the target host"
  fi
  if [[ "$sync_try_to_pressure_nodes_flag" == "true" ]]; then
    sync_try_to_pressure_nodes "$target_manifest" "$try_root" "true"
  fi
  if [[ "$dry_run" == "true" ]]; then
    echo "dry_run=true"
  else
    if [[ "$sync_try_to_pressure_nodes_flag" != "true" ]]; then
      sync_try_to_pressure_nodes "$target_manifest" "$try_root" "false"
    fi
		if [[ "$pre_run_recovery_required" == "true" ]]; then
			if "$tipoc_bin" run recover --manifest "$target_manifest"; then
				jq '.pre_run_recovery.status = "passed"' "$receipt" > "${receipt}.tmp" && mv "${receipt}.tmp" "$receipt"
			else
				code=$?
				jq --argjson code "$code" '.status = "failed" | .exit_code = $code | .pre_run_recovery.status = "failed"' "$receipt" > "${receipt}.tmp" && mv "${receipt}.tmp" "$receipt"
				exit "$code"
			fi
		fi
    run_args=(run --manifest "$target_manifest")
    if [[ -n "$operation_runtime_overrides" ]]; then
      run_args+=(--operation-runtime-overrides "$operation_runtime_overrides")
    fi
    if "$tipoc_bin" "${run_args[@]}"; then
      jq '.status = "passed" | .exit_code = 0' "$receipt" > "${receipt}.tmp" && mv "${receipt}.tmp" "$receipt"
    else
      code=$?
      jq --argjson code "$code" '.status = "failed" | .exit_code = $code' "$receipt" > "${receipt}.tmp" && mv "${receipt}.tmp" "$receipt"
      exit "$code"
    fi
  fi
done