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

readonly DISPATCHER_VERSION="teleo-gcp-iap-operator-v1"
readonly EXPECTED_WORKFLOW_REF="living-ip/teleo-infrastructure/.github/workflows/gcp-iap-operator.yml@refs/heads/main"
readonly PROJECT_ID="teleo-501523"
readonly INSTANCE="teleo-prod-1"
readonly ZONE="europe-west6-a"
readonly INSTANCE_INTERNAL_IP="10.60.0.3"
readonly CLOUDSQL_HOST="10.61.0.3"
readonly CLOUDSQL_SECRET="gcp-teleo-pgvector-standby-postgres-password"
readonly REMOTE_DISPATCHER="/usr/local/sbin/teleo-gcp-iap-operator"
readonly REMOTE_RUN_ROOT="/var/lib/teleo-gcp-iap-operator/runs"
readonly LIVE_PROFILE="/home/teleo/.hermes/profiles/leoclean"
readonly LIVE_SERVICE="leoclean-gcp-prod-parallel.service"
readonly STATUS_TARGET="teleo_clone_status"
readonly REQUEST_ID_RE='^iap-[a-z0-9]{12,32}$'
readonly TARGET_DB_RE='^teleo_clone_[a-z0-9][a-z0-9_]{0,50}$'

RUNNER_TEMP_DIR=""
VALIDATED_BUNDLE_DIR=""
VALIDATED_BUNDLE_SHA256=""
VALIDATED_BUNDLE_COMMIT=""

die() {
  printf 'gcp_iap_operator: %s\n' "$1" >&2
  exit 2
}

cleanup_runner_temp() {
  if [[ -n "$RUNNER_TEMP_DIR" && -d "$RUNNER_TEMP_DIR" && "$RUNNER_TEMP_DIR" == "${RUNNER_TEMP:-}/"* ]]; then
    rm -rf -- "$RUNNER_TEMP_DIR"
  fi
}

validate_inputs() {
  local operation="$1"
  local request_id="$2"
  local target_db="$3"

  case "$operation" in
    status | direct-claim-replay | cleanup-clone) ;;
    *) die "operation is not allowlisted" ;;
  esac
  [[ "$request_id" =~ $REQUEST_ID_RE ]] || die "request_id does not match the required format"
  [[ "$target_db" =~ $TARGET_DB_RE ]] || die "target_db must be a bounded teleo_clone_* identifier"
  if [[ "$operation" == "status" && "$target_db" != "$STATUS_TARGET" ]]; then
    die "status requires the fixed teleo_clone_status sentinel"
  fi
  if [[ "$operation" != "status" && "$target_db" == "$STATUS_TARGET" ]]; then
    die "clone operations cannot use the status sentinel"
  fi
}

emit_json() {
  local operation="$1"
  local request_id="$2"
  local target_db="$3"
  local status="$4"
  shift 4
  python3 - "$operation" "$request_id" "$target_db" "$status" "$@" <<'PY'
import json
import sys

operation, request_id, target_db, status, *pairs = sys.argv[1:]
payload = {
    "schema": "livingip.gcpIapOperatorRemoteResult.v1",
    "operation": operation,
    "request_id": request_id,
    "target_db": target_db,
    "status": status,
}
for pair in pairs:
    key, separator, value = pair.partition("=")
    if not separator:
        raise SystemExit("invalid result field")
    if key in {"nrestarts", "result_bytes"}:
        payload[key] = int(value)
    elif key in {"no_message_send", "live_service_unchanged", "live_profile_unchanged"}:
        payload[key] = value == "true"
    else:
        payload[key] = value
print(json.dumps(payload, sort_keys=True))
PY
}

remote_status() {
  local request_id="$1"
  local target_db="$2"
  local active_state sub_state nrestarts dispatcher_hash dispatcher_state

  [[ -x "$REMOTE_DISPATCHER" && ! -L "$REMOTE_DISPATCHER" ]] || die "root-owned remote dispatcher is absent"
  dispatcher_state="$(stat -c '%U:%G:%a' "$REMOTE_DISPATCHER")"
  [[ "$dispatcher_state" == "root:root:755" ]] || die "remote dispatcher ownership or mode drifted"
  dispatcher_hash="$(sha256sum "$REMOTE_DISPATCHER" | awk '{print $1}')"
  [[ "$dispatcher_hash" =~ ^[0-9a-f]{64}$ ]] || die "remote dispatcher hash readback was invalid"
  active_state="$(systemctl show "$LIVE_SERVICE" --property=ActiveState --value)"
  sub_state="$(systemctl show "$LIVE_SERVICE" --property=SubState --value)"
  nrestarts="$(systemctl show "$LIVE_SERVICE" --property=NRestarts --value)"
  [[ "$active_state" =~ ^[a-z]+$ ]] || die "service ActiveState readback was invalid"
  [[ "$sub_state" =~ ^[a-z-]+$ ]] || die "service SubState readback was invalid"
  [[ "$nrestarts" =~ ^[0-9]+$ ]] || die "service NRestarts readback was invalid"

  emit_json \
    "status" "$request_id" "$target_db" "pass" \
    "instance=$INSTANCE" \
    "expected_internal_ip=$INSTANCE_INTERNAL_IP" \
    "active_state=$active_state" \
    "sub_state=$sub_state" \
    "nrestarts=$nrestarts" \
    "dispatcher_version=$DISPATCHER_VERSION" \
    "dispatcher_sha256=$dispatcher_hash"
}

require_root() {
  [[ "$EUID" -eq 0 ]] || die "clone operations require the isolated OS Admin identity"
  [[ -n "${SUDO_USER:-}" && "$SUDO_USER" != "root" ]] || die "clone operation lacks a distinct OS Login caller"
}

parity_receipt_relative() {
  local target_db="$1"
  printf 'docs/reports/leo-working-state-20260709/%s-canonical-parity-receipt.json\n' "$target_db"
}

write_run_marker() {
  local marker="$1"
  local request_id="$2"
  local target_db="$3"
  printf 'request_id=%s\ntarget_db=%s\n' "$request_id" "$target_db" >"$marker"
  chmod 0600 "$marker"
}

validate_run_dir() {
  local run_dir="$1"
  local request_id="$2"
  local target_db="$3"
  local marker="$run_dir/.run-owner"
  local resolved expected

  [[ -d "$run_dir" && ! -L "$run_dir" ]] || die "exact run-owned directory is absent or unsafe"
  resolved="$(realpath "$run_dir")"
  expected="$REMOTE_RUN_ROOT/$request_id"
  [[ "$resolved" == "$expected" ]] || die "run-owned directory escaped its fixed root"
  [[ "$(stat -c '%U:%G:%a' "$run_dir")" == "root:root:700" ]] || die "run directory ownership or mode drifted"
  [[ -f "$marker" && ! -L "$marker" ]] || die "run ownership marker is absent or unsafe"
  [[ "$(cat "$marker")" == $'request_id='"$request_id"$'\ntarget_db='"$target_db" ]] || {
    die "run ownership marker does not match request_id and target_db"
  }
}

remote_upload_path() {
  local request_id="$1"
  local passwd_entry home resolved upload

  passwd_entry="$(getent passwd "$SUDO_USER")"
  [[ -n "$passwd_entry" ]] || die "OS Login caller has no passwd entry"
  home="$(cut -d: -f6 <<<"$passwd_entry")"
  [[ "$home" == /* && -d "$home" && ! -L "$home" ]] || die "OS Login caller home is unsafe"
  resolved="$(realpath "$home")"
  [[ "$resolved" == "$home" ]] || die "OS Login caller home path drifted"
  upload="$home/.teleo-iap-upload-$request_id.tar.gz"
  [[ -f "$upload" && ! -L "$upload" ]] || die "request-derived bundle upload is absent or unsafe"
  [[ "$(stat -c '%U:%a' "$upload")" == "$SUDO_USER:600" ]] || die "bundle upload owner or mode is unsafe"
  printf '%s\n' "$upload"
}

validate_and_extract_bundle() {
  local archive="$1"
  local bundle_dir="$2"
  local operation="$3"
  local request_id="$4"
  local target_db="$5"

  python3 - \
    "$archive" "$bundle_dir" "$operation" "$request_id" "$target_db" \
    "$REMOTE_DISPATCHER" "$DISPATCHER_VERSION" "$EXPECTED_WORKFLOW_REF" <<'PY'
import hashlib
import json
import os
import re
import sys
import tarfile
from pathlib import Path

archive_path = Path(sys.argv[1])
bundle_dir = Path(sys.argv[2]).resolve()
operation, request_id, target_db = sys.argv[3:6]
dispatcher_path = Path(sys.argv[6])
dispatcher_version = sys.argv[7]
expected_workflow_ref = sys.argv[8]
common = ["scripts/gcp_iap_operator.sh"]
direct_claim = [
    "scripts/run_gcp_generated_db_direct_claim_suite.py",
    "scripts/run_leo_clone_bound_handler_checkpoint.py",
    "scripts/working_leo_open_ended_benchmark.py",
    "hermes-agent/leoclean-bin/cloudsql_memory_tool.py",
    "ops/postgres_parity_manifest.sql",
    f"docs/reports/leo-working-state-20260709/{target_db}-canonical-parity-receipt.json",
]
expected = common + (direct_claim if operation == "direct-claim-replay" else [])
if operation not in {"direct-claim-replay", "cleanup-clone"}:
    raise SystemExit("unexpected bundle operation")
if archive_path.stat().st_size > 5 * 1024 * 1024:
    raise SystemExit("bundle archive exceeds size ceiling")

with tarfile.open(archive_path, mode="r:gz") as archive:
    members = archive.getmembers()
    expected_members = sorted([*expected, "bundle-manifest.json"])
    names = [member.name for member in members]
    if sorted(names) != expected_members or len(names) != len(set(names)):
        raise SystemExit("bundle member allowlist mismatch")
    if any(not member.isfile() or member.size > 2 * 1024 * 1024 for member in members):
        raise SystemExit("bundle contains a link, special entry, or oversized file")
    manifest_member = archive.getmember("bundle-manifest.json")
    if manifest_member.size > 64 * 1024:
        raise SystemExit("bundle manifest exceeds size ceiling")
    manifest = json.loads(archive.extractfile(manifest_member).read())
    required = {
        "schema": "livingip.gcpIapOperatorBundle.v1",
        "dispatcher_version": dispatcher_version,
        "operation": operation,
        "request_id": request_id,
        "target_db": target_db,
        "ref": "refs/heads/main",
        "workflow_ref": expected_workflow_ref,
    }
    if any(manifest.get(key) != value for key, value in required.items()):
        raise SystemExit("bundle manifest context mismatch")
    if not re.fullmatch(r"[0-9a-f]{40}", str(manifest.get("git_commit") or "")):
        raise SystemExit("bundle manifest commit is invalid")
    files = manifest.get("files")
    if not isinstance(files, dict) or sorted(files) != sorted(expected):
        raise SystemExit("bundle manifest file allowlist mismatch")
    payloads = {}
    for relative in expected:
        member = archive.getmember(relative)
        data = archive.extractfile(member).read()
        receipt = files.get(relative) or {}
        if receipt.get("size") != len(data) or receipt.get("sha256") != hashlib.sha256(data).hexdigest():
            raise SystemExit(f"bundle hash mismatch: {relative}")
        payloads[relative] = data
    dispatcher_sha = hashlib.sha256(dispatcher_path.read_bytes()).hexdigest()
    if dispatcher_sha != files["scripts/gcp_iap_operator.sh"]["sha256"]:
        raise SystemExit("installed dispatcher does not match the reviewed main bundle")
    manifest_bytes = archive.extractfile(manifest_member).read()

for relative, data in sorted({**payloads, "bundle-manifest.json": manifest_bytes}.items()):
    destination = (bundle_dir / relative).resolve()
    if bundle_dir not in destination.parents:
        raise SystemExit("bundle destination escaped the run directory")
    destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    os.chmod(destination.parent, 0o700)
    with destination.open("xb") as handle:
        handle.write(data)
    os.chmod(destination, 0o600)
PY
}

receive_and_validate_bundle() {
  local operation="$1"
  local request_id="$2"
  local target_db="$3"
  local upload run_dir archive bundle_dir

  require_root
  upload="$(remote_upload_path "$request_id")"
  run_dir="$REMOTE_RUN_ROOT/$request_id"
  if [[ "$operation" == "direct-claim-replay" ]]; then
    [[ ! -e "$run_dir" ]] || die "request_id already owns a remote run directory"
    install -d -o root -g root -m 0700 "$run_dir"
    write_run_marker "$run_dir/.run-owner" "$request_id" "$target_db"
  else
    validate_run_dir "$run_dir" "$request_id" "$target_db"
  fi
  archive="$run_dir/$operation-bundle.tar.gz"
  bundle_dir="$run_dir/$operation-bundle"
  [[ ! -e "$archive" && ! -e "$bundle_dir" ]] || die "operation bundle already exists in run directory"
  mv -- "$upload" "$archive"
  chown root:root "$archive"
  chmod 0600 "$archive"
  install -d -o root -g root -m 0700 "$bundle_dir"
  validate_and_extract_bundle "$archive" "$bundle_dir" "$operation" "$request_id" "$target_db" || {
    die "bundle membership, hash, context, or dispatcher validation failed"
  }
  VALIDATED_BUNDLE_DIR="$bundle_dir"
  VALIDATED_BUNDLE_SHA256="$(sha256sum "$archive" | awk '{print $1}')"
  VALIDATED_BUNDLE_COMMIT="$(
    python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["git_commit"])' \
      "$bundle_dir/bundle-manifest.json"
  )"
}

remote_direct_claim_replay() {
  local request_id="$1"
  local target_db="$2"
  local receipt_relative receipt_path run_dir output_path stdout_path stderr_path report_sha result_bytes

  receive_and_validate_bundle "direct-claim-replay" "$request_id" "$target_db"
  receipt_relative="$(parity_receipt_relative "$target_db")"
  receipt_path="$VALIDATED_BUNDLE_DIR/$receipt_relative"
  [[ -s "$receipt_path" && ! -L "$receipt_path" ]] || die "target-specific tracked parity receipt is absent"
  [[ -d "$LIVE_PROFILE" ]] || die "live profile required for temporary copy is absent"

  run_dir="$REMOTE_RUN_ROOT/$request_id"
  output_path="$run_dir/direct-claim-result.json"
  stdout_path="$run_dir/direct-claim.stdout"
  stderr_path="$run_dir/direct-claim.stderr"
  : >"$stdout_path"
  : >"$stderr_path"
  chmod 0600 "$stdout_path" "$stderr_path"

  if ! (
    cd "$VALIDATED_BUNDLE_DIR"
    /usr/bin/python3 scripts/run_gcp_generated_db_direct_claim_suite.py \
      --execute \
      --target-db "$target_db" \
      --cloudsql-tool hermes-agent/leoclean-bin/cloudsql_memory_tool.py \
      --manifest-sql ops/postgres_parity_manifest.sql \
      --parity-receipt "$receipt_path" \
      --output "$output_path" \
      --host "$CLOUDSQL_HOST" \
      --project "$PROJECT_ID" \
      --password-secret "$CLOUDSQL_SECRET" \
      --service "$LIVE_SERVICE" \
      --live-profile "$LIVE_PROFILE" \
      --run-id "$request_id"
  ) >"$stdout_path" 2>"$stderr_path"; then
    die "reviewed six-turn clone-bound replay failed; private run files were retained for cleanup"
  fi

  [[ -s "$output_path" && ! -L "$output_path" ]] || die "six-turn replay did not retain its private receipt"
  python3 - "$output_path" "$target_db" <<'PY'
import json
import sys
from pathlib import Path

payload = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))
target_db = sys.argv[2]
checks = payload.get("checks") or {}
required = {
    "status": payload.get("status") == "pass",
    "target": payload.get("target_database") == target_db,
    "six_replies": checks.get("six_replies_returned") is True,
    "no_message_send": payload.get("posted_to_telegram") is False,
    "service_unchanged": checks.get("live_service_unchanged") is True,
    "profile_unchanged": checks.get("live_profile_unchanged") is True,
    "database_unchanged": checks.get("generated_database_unchanged") is True,
    "temporary_profile_removed": checks.get("temporary_profile_removed") is True,
}
failed = sorted(name for name, ok in required.items() if not ok)
if failed:
    raise SystemExit("replay receipt failed fixed checks: " + ", ".join(failed))
PY
  chmod 0600 "$output_path"
  report_sha="$(sha256sum "$output_path" | awk '{print $1}')"
  result_bytes="$(wc -c <"$output_path" | tr -d ' ')"
  emit_json \
    "direct-claim-replay" "$request_id" "$target_db" "pass" \
    "bundle_commit=$VALIDATED_BUNDLE_COMMIT" \
    "bundle_sha256=$VALIDATED_BUNDLE_SHA256" \
    "result_sha256=$report_sha" \
    "result_bytes=$result_bytes" \
    "no_message_send=true" \
    "live_service_unchanged=true" \
    "live_profile_unchanged=true"
}

remote_cleanup_clone() {
  local request_id="$1"
  local target_db="$2"
  local run_dir password exists remaining

  receive_and_validate_bundle "cleanup-clone" "$request_id" "$target_db"
  run_dir="$REMOTE_RUN_ROOT/$request_id"
  validate_run_dir "$run_dir" "$request_id" "$target_db"
  command -v gcloud >/dev/null 2>&1 || die "gcloud is unavailable for secret-backed clone cleanup"
  command -v psql >/dev/null 2>&1 || die "psql is unavailable for clone cleanup"

  password="$(gcloud secrets versions access latest --secret="$CLOUDSQL_SECRET" --project="$PROJECT_ID" --quiet)"
  [[ -n "$password" ]] || die "clone cleanup credential resolved empty"
  export PGPASSWORD="$password"
  export PGOPTIONS="-c statement_timeout=30000"
  exists="$(
    psql "host=$CLOUDSQL_HOST port=5432 dbname=postgres user=postgres sslmode=require connect_timeout=8" \
      -X -Atq -v ON_ERROR_STOP=1 -v "target_db=$target_db" \
      -c "select exists(select 1 from pg_database where datname = :'target_db');"
  )"
  [[ "$exists" == "t" ]] || die "marked clone database is absent; refusing directory-only cleanup"

  psql "host=$CLOUDSQL_HOST port=5432 dbname=postgres user=postgres sslmode=require connect_timeout=8" \
    -X -q -v ON_ERROR_STOP=1 -v "target_db=$target_db" <<'SQL'
select pg_terminate_backend(pid)
from pg_stat_activity
where datname = :'target_db' and pid <> pg_backend_pid();
drop database :"target_db";
SQL
  remaining="$(
    psql "host=$CLOUDSQL_HOST port=5432 dbname=postgres user=postgres sslmode=require connect_timeout=8" \
      -X -Atq -v ON_ERROR_STOP=1 -v "target_db=$target_db" \
      -c "select count(*) from pg_database where datname = :'target_db';"
  )"
  unset PGPASSWORD PGOPTIONS
  password=""
  [[ "$remaining" == "0" ]] || die "clone database remained after cleanup"

  validate_run_dir "$run_dir" "$request_id" "$target_db"
  rm -rf --one-file-system -- "$run_dir"
  [[ ! -e "$run_dir" ]] || die "exact run-owned directory remained after cleanup"
  emit_json \
    "cleanup-clone" "$request_id" "$target_db" "pass" \
    "bundle_commit=$VALIDATED_BUNDLE_COMMIT" \
    "bundle_sha256=$VALIDATED_BUNDLE_SHA256" \
    "clone_database_remaining=0"
}

remote_main() {
  [[ "$#" -eq 3 ]] || die "remote mode requires operation, request_id, and target_db"
  local operation="$1"
  local request_id="$2"
  local target_db="$3"
  validate_inputs "$operation" "$request_id" "$target_db"
  case "$operation" in
    status) remote_status "$request_id" "$target_db" ;;
    direct-claim-replay) remote_direct_claim_replay "$request_id" "$target_db" ;;
    cleanup-clone) remote_cleanup_clone "$request_id" "$target_db" ;;
  esac
}

write_sanitized_result() {
  local output_path="$1"
  local raw_stdout="$2"
  local operation="$3"
  local request_id="$4"
  local target_db="$5"
  local exit_code="$6"
  RESULT_PATH="$output_path" RAW_STDOUT="$raw_stdout" OPERATION="$operation" \
    REQUEST_ID="$request_id" TARGET_DB="$target_db" EXIT_CODE="$exit_code" python3 - <<'PY'
import hashlib
import json
import os
from pathlib import Path

raw = Path(os.environ["RAW_STDOUT"]).read_bytes()
exit_code = int(os.environ["EXIT_CODE"])
remote = {}
if exit_code == 0:
    try:
        candidate = json.loads(raw.decode("utf-8").splitlines()[-1])
        allowed = {
            "operation",
            "request_id",
            "target_db",
            "status",
            "instance",
            "expected_internal_ip",
            "active_state",
            "sub_state",
            "nrestarts",
            "dispatcher_version",
            "dispatcher_sha256",
            "bundle_commit",
            "bundle_sha256",
            "result_sha256",
            "result_bytes",
            "no_message_send",
            "live_service_unchanged",
            "live_profile_unchanged",
            "clone_database_remaining",
        }
        remote = {key: value for key, value in candidate.items() if key in allowed}
    except (UnicodeDecodeError, json.JSONDecodeError, IndexError):
        exit_code = 90

payload = {
    "schema": "livingip.gcpIapOperatorSanitizedResult.v1",
    "operation": os.environ["OPERATION"],
    "request_id": os.environ["REQUEST_ID"],
    "target_db": os.environ["TARGET_DB"],
    "status": "pass" if exit_code == 0 and remote.get("status") == "pass" else "fail",
    "ssh_exit_code": exit_code,
    "remote_result": remote,
    "remote_stdout_sha256": hashlib.sha256(raw).hexdigest(),
    "remote_stdout_bytes": len(raw),
    "raw_stdout_retained": False,
    "raw_stderr_retained": False,
    "credential_values_logged": False,
    "ssh_debug_enabled": False,
}
Path(os.environ["RESULT_PATH"]).write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
PY
  chmod 0600 "$output_path"
}

validate_local_bundle() {
  local request_id="$1"
  local expected path resolved bundle_root mode

  bundle_root="$RUNNER_TEMP/gcp-iap-operator-bundle"
  expected="$bundle_root/$request_id.tar.gz"
  path="$expected"
  [[ -f "$path" && ! -L "$path" ]] || die "fixed workflow bundle is absent or unsafe"
  resolved="$(realpath "$path")"
  [[ "$resolved" == "$expected" ]] || die "workflow bundle escaped its request-derived path"
  mode="$(python3 -c 'import os,stat,sys; print(oct(stat.S_IMODE(os.stat(sys.argv[1]).st_mode))[2:])' "$path")"
  [[ "$mode" == "600" ]] || die "workflow bundle mode is not private"
  printf '%s\n' "$path"
}

runner_main() {
  [[ "$#" -eq 3 ]] || die "runner mode requires operation, request_id, and target_db"
  local operation="$1"
  local request_id="$2"
  local target_db="$3"
  local temp_dir result_dir result_path raw_stdout raw_stderr scp_stdout ssh_key remote_command exit_code
  local sanitized_status bundle_path remote_upload
  local -a remote_argv

  validate_inputs "$operation" "$request_id" "$target_db"
  [[ -n "${RUNNER_TEMP:-}" && "$RUNNER_TEMP" == /* ]] || die "RUNNER_TEMP must be an absolute runner path"
  [[ -n "${RESULT_DIR:-}" && "$RESULT_DIR" == "$RUNNER_TEMP"/* ]] || {
    die "RESULT_DIR must be a fixed child of RUNNER_TEMP"
  }
  command -v gcloud >/dev/null 2>&1 || die "gcloud is unavailable"
  install -d -m 0700 "$RESULT_DIR"
  result_dir="$(realpath "$RESULT_DIR")"
  [[ "$result_dir" == "$RUNNER_TEMP"/* ]] || die "RESULT_DIR escaped RUNNER_TEMP"
  result_path="$result_dir/result.json"
  : >"$result_path"
  chmod 0600 "$result_path"

  temp_dir="$(mktemp -d "$RUNNER_TEMP/teleo-iap-operator.XXXXXX")"
  RUNNER_TEMP_DIR="$temp_dir"
  chmod 0700 "$temp_dir"
  trap cleanup_runner_temp EXIT
  trap 'exit 129' HUP
  trap 'exit 130' INT
  trap 'exit 143' TERM
  raw_stdout="$temp_dir/remote.stdout"
  raw_stderr="$temp_dir/transport.stderr"
  scp_stdout="$temp_dir/scp.stdout"
  ssh_key="$temp_dir/ssh_key"
  : >"$raw_stdout"
  : >"$raw_stderr"
  : >"$scp_stdout"
  chmod 0600 "$raw_stdout" "$raw_stderr" "$scp_stdout"

  if [[ "$operation" != "status" ]]; then
    bundle_path="$(validate_local_bundle "$request_id")"
    remote_upload="$INSTANCE:.teleo-iap-upload-$request_id.tar.gz"
    set +e
    gcloud compute scp "$bundle_path" "$remote_upload" \
      --project="$PROJECT_ID" \
      --zone="$ZONE" \
      --tunnel-through-iap \
      --ssh-key-file="$ssh_key" \
      --ssh-key-expire-after=5m \
      --scp-flag=-p \
      --quiet >"$scp_stdout" 2>>"$raw_stderr"
    exit_code=$?
    set -e
    if [[ "$exit_code" -ne 0 ]]; then
      write_sanitized_result "$result_path" "$raw_stdout" "$operation" "$request_id" "$target_db" "$exit_code"
      printf 'sanitized_result=%s\n' "$result_path"
      return "$exit_code"
    fi
  fi

  if [[ "$operation" == "status" ]]; then
    remote_argv=("$REMOTE_DISPATCHER" --remote "$operation" "$request_id" "$target_db")
  else
    remote_argv=(sudo -n "$REMOTE_DISPATCHER" --remote "$operation" "$request_id" "$target_db")
  fi
  printf -v remote_command '%q ' "${remote_argv[@]}"

  set +e
  gcloud compute ssh "$INSTANCE" \
    --project="$PROJECT_ID" \
    --zone="$ZONE" \
    --tunnel-through-iap \
    --ssh-key-file="$ssh_key" \
    --ssh-key-expire-after=5m \
    --command="$remote_command" \
    --quiet >"$raw_stdout" 2>>"$raw_stderr"
  exit_code=$?
  set -e
  [[ ! -e "$ssh_key" ]] || chmod 0600 "$ssh_key"
  [[ ! -e "$ssh_key.pub" ]] || chmod 0600 "$ssh_key.pub"
  write_sanitized_result "$result_path" "$raw_stdout" "$operation" "$request_id" "$target_db" "$exit_code"
  sanitized_status="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["status"])' "$result_path")"
  if [[ "$exit_code" -eq 0 && "$sanitized_status" != "pass" ]]; then
    exit_code=90
  fi
  printf 'sanitized_result=%s\n' "$result_path"
  return "$exit_code"
}

if [[ "${1:-}" == "--remote" ]]; then
  shift
  remote_main "$@"
else
  runner_main "$@"
fi