#!/usr/bin/env bash
# Fixed Host110 executor for Windows99 Agent99 backup-runtime deployment.

set -euo pipefail
umask 077

readonly EXPECTED_HOST_IP="192.168.0.110"
readonly EXPECTED_USER="wooo"
readonly DEST_ROOT="/backup/scripts"
readonly EXPORTER_ROOT="/home/wooo/scripts"
readonly EXPORTER_FILE="backup-health-textfile-exporter.py"
readonly STATUS_ROOT="/backup/status"
readonly STAGE_ROOT="/backup/.agent99-backup-runtime-stage"
readonly ROLLBACK_ROOT="/backup/.agent99-backup-runtime-rollback"
readonly LOCK_PATH="/tmp/agent99-host110-backup-runtime.lock"

readonly -a RUNTIME_FILES=(
  common.sh
  backup-all.sh
  backup-host188-products.sh
  verify-host188-products-backup.sh
  verify-host188-products-archive.py
  backup-gitea.sh
  gitea-full-backup-restore-drill.sh
  backup-configs.sh
  backup-awoooi.sh
  backup-awoooi-frequent.sh
  backup-clawbot.sh
  backup-sentry.sh
  check-backup-integrity.sh
  sync-offsite-backups.sh
  backup-offsite-readiness-gate.sh
  verify-offsite-full-sync.sh
  enforce-latest-only-retention.sh
)
readonly -a PAYLOAD_FILES=("${RUNTIME_FILES[@]}" "$EXPORTER_FILE")

MODE=""
SOURCE_REVISION=""
RUN_ID=""
SOURCE_STAGE=""
SOURCE_HEAD=""
EXECUTOR_DIGEST=""
VERIFIER_DIGEST=""
STAGE_DIR=""
ROLLBACK_DIR=""
RECEIPT_PATH=""
APPLY_STARTED=0
APPLY_COMPLETE=0
ROLLBACK_ATTEMPTED=0
ROLLBACK_PERFORMED=0
ROLLBACK_VERIFIED=0
RUN_TEMP_RESIDUE_VERIFIED=0
FAULT_INJECTION_ENABLED=0
declare -A FILE_EXISTED=()
declare -A PREVIOUS_DIGEST=()
declare -A PREVIOUS_METADATA=()
declare -A EXPECTED_DIGEST=()
declare -A RUN_OWNED_TEMP_PATHS=()

usage() {
  printf '%s\n' \
    "Usage: $0 --mode check|apply|verify --source-revision SHA --run-id ID --source-stage PATH" \
    "The executor is fixed to host 192.168.0.110, an Agent99 run-bound source stage," \
    "and the 18-file transaction: 17 /backup/scripts files plus the backup health exporter."
}

fail() {
  printf 'HOST110_BACKUP_RUNTIME_OK=0\nERROR=%s\n' "$1" >&2
  exit 64
}

while [ "$#" -gt 0 ]; do
  case "$1" in
    --mode)
      shift
      MODE="${1:-}"
      ;;
    --source-revision)
      shift
      SOURCE_REVISION="${1:-}"
      ;;
    --run-id)
      shift
      RUN_ID="${1:-}"
      ;;
    --source-stage)
      shift
      SOURCE_STAGE="${1:-}"
      ;;
    -h|--help)
      usage
      exit 0
      ;;
    *) fail "unknown_argument" ;;
  esac
  shift
done

case "$MODE" in check|apply|verify) ;; *) fail "invalid_mode" ;; esac
[[ "$SOURCE_REVISION" =~ ^[0-9a-f]{40}$ ]] || fail "invalid_source_revision"
[[ "$RUN_ID" =~ ^[A-Za-z0-9][A-Za-z0-9._-]{0,95}$ ]] || fail "invalid_run_id"
expected_source_stage="/tmp/agent99-host110-backup-runtime-${RUN_ID}"
[ "$SOURCE_STAGE" = "$expected_source_stage" ] || fail "invalid_source_stage"
RECEIPT_PATH="$STATUS_ROOT/agent99-backup-runtime-deploy-${RUN_ID}.json"
if [ "${HOST110_BACKUP_RUNTIME_ENABLE_TEST_FAULTS:-}" = "1" ] \
  && [ "$DEST_ROOT" != "/backup/scripts" ]; then
  FAULT_INJECTION_ENABLED=1
fi

for command_name in awk bash cp dirname flock grep hostname id install mv pgrep python3 readlink rm sha256sum stat timeout tr; do
  command -v "$command_name" >/dev/null 2>&1 || fail "required_command_missing_${command_name}"
done

current_user="$(id -un)"
[ "$current_user" = "$EXPECTED_USER" ] || fail "unexpected_executor_user"
hostname -I 2>/dev/null | tr ' ' '\n' | grep -qx "$EXPECTED_HOST_IP" || fail "unexpected_executor_host"
[ -d "$SOURCE_STAGE" ] && [ ! -L "$SOURCE_STAGE" ] || fail "canonical_source_stage_unavailable"
[ -d "$DEST_ROOT" ] && [ ! -L "$DEST_ROOT" ] || fail "backup_runtime_destination_unavailable"
[ -d "$EXPORTER_ROOT" ] && [ ! -L "$EXPORTER_ROOT" ] || fail "backup_exporter_destination_unavailable"

manifest_rows="$(python3 - "$SOURCE_STAGE/manifest.json" "$SOURCE_REVISION" "$RUN_ID" "${PAYLOAD_FILES[@]}" <<'PY'
import json
import re
import sys
from pathlib import Path

path = Path(sys.argv[1])
source_revision = sys.argv[2]
run_id = sys.argv[3]
expected_names = sys.argv[4:]
try:
    document = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
    raise SystemExit(65)
rows = document.get("files")
if (
    document.get("schemaVersion") != "agent99_host110_backup_runtime_source_manifest_v1"
    or document.get("sourceRevision") != source_revision
    or document.get("runId") != run_id
    or not re.fullmatch(r"[0-9a-f]{40}", str(document.get("sourceHead", "")))
    or not re.fullmatch(r"[0-9a-f]{64}", str(document.get("executorSha256", "")))
    or not re.fullmatch(r"[0-9a-f]{64}", str(document.get("verifierSha256", "")))
    or not isinstance(rows, list)
    or len(rows) != len(expected_names)
):
    raise SystemExit(66)
by_name = {}
for row in rows:
    if not isinstance(row, dict):
        raise SystemExit(66)
    name = str(row.get("name", ""))
    digest = str(row.get("sha256", ""))
    if name in by_name or not re.fullmatch(r"[A-Za-z0-9_.-]+", name) or not re.fullmatch(r"[0-9a-f]{64}", digest):
        raise SystemExit(66)
    by_name[name] = digest
if set(by_name) != set(expected_names):
    raise SystemExit(66)
print(f"SOURCE_HEAD {document['sourceHead']}")
print(f"EXECUTOR_SHA256 {document['executorSha256']}")
print(f"VERIFIER_SHA256 {document['verifierSha256']}")
for name in expected_names:
    print(f"{name} {by_name[name]}")
PY
)" || fail "source_manifest_invalid"

while read -r manifest_name manifest_digest; do
  case "$manifest_name" in
    SOURCE_HEAD) SOURCE_HEAD="$manifest_digest" ;;
    EXECUTOR_SHA256) EXECUTOR_DIGEST="$manifest_digest" ;;
    VERIFIER_SHA256) VERIFIER_DIGEST="$manifest_digest" ;;
    *) EXPECTED_DIGEST[$manifest_name]="$manifest_digest" ;;
  esac
done <<< "$manifest_rows"
[[ "$SOURCE_HEAD" =~ ^[0-9a-f]{40}$ ]] || fail "source_head_invalid"

working_digest() {
  sha256sum "$1" | awk '{print $1}'
}

[ "$(working_digest "$0")" = "$EXECUTOR_DIGEST" ] || fail "executor_identity_failed"
[ -f "$SOURCE_STAGE/verify-host110-backup-runtime.py" ] \
  && [ ! -L "$SOURCE_STAGE/verify-host110-backup-runtime.py" ] \
  && [ "$(working_digest "$SOURCE_STAGE/verify-host110-backup-runtime.py")" = "$VERIFIER_DIGEST" ] \
  || fail "verifier_identity_failed"

validate_file() {
  local validation_path="$1"
  case "$validation_path" in
    *.sh) bash -n "$validation_path" ;;
    *.py) python3 -c 'import ast, pathlib, sys; ast.parse(pathlib.Path(sys.argv[1]).read_text(encoding="utf-8"))' "$validation_path" ;;
    *) return 64 ;;
  esac
}

source_digest() {
  local name="$1"
  printf '%s\n' "${EXPECTED_DIGEST[$name]:-}"
}

validate_source() {
  local name source_path expected_digest actual_digest resolved
  for name in "${PAYLOAD_FILES[@]}"; do
    source_path="$SOURCE_STAGE/$name"
    [ -f "$source_path" ] && [ ! -L "$source_path" ] || return 65
    resolved="$(readlink -f "$source_path")"
    [[ "$resolved" == "$SOURCE_STAGE/"* ]] || return 65
    expected_digest="$(source_digest "$name")"
    actual_digest="$(working_digest "$source_path")"
    [ "$actual_digest" = "$expected_digest" ] || return 66
    validate_file "$source_path" || return 67
  done
}

destination_path() {
  local name="$1"
  if [ "$name" = "$EXPORTER_FILE" ]; then
    printf '%s/%s\n' "$EXPORTER_ROOT" "$name"
  else
    printf '%s/%s\n' "$DEST_ROOT" "$name"
  fi
}

verify_destination() {
  local name dest_path expected_digest actual_digest
  for name in "${PAYLOAD_FILES[@]}"; do
    dest_path="$(destination_path "$name")"
    [ -f "$dest_path" ] && [ ! -L "$dest_path" ] || return 68
    expected_digest="$(source_digest "$name")"
    actual_digest="$(working_digest "$dest_path")"
    [ "$actual_digest" = "$expected_digest" ] || return 69
    [ "$(stat -c '%U:%G:%a' "$dest_path")" = "wooo:wooo:755" ] || return 70
  done
}

fsync_paths_and_parents() {
  python3 - "$@" <<'PY'
import os
import stat
import sys
from pathlib import Path

parents = set()
for raw_path in sys.argv[1:]:
    path = Path(raw_path)
    parents.add(path.parent)
    if not os.path.lexists(path):
        continue
    if path.is_symlink() or not path.is_file():
        raise SystemExit(76)
    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags)
    try:
        if not stat.S_ISREG(os.fstat(descriptor).st_mode):
            raise SystemExit(76)
        os.fsync(descriptor)
    finally:
        os.close(descriptor)
for parent in sorted(parents, key=str):
    if parent.is_symlink() or not parent.is_dir():
        raise SystemExit(76)
    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(parent, flags)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)
PY
}

fsync_destination_state() {
  local name dest_path
  local -a fsync_targets=()
  for name in "${PAYLOAD_FILES[@]}"; do
    dest_path="$(destination_path "$name")"
    fsync_targets+=("$dest_path")
  done
  fsync_paths_and_parents "${fsync_targets[@]}"
}

fsync_rollback_prestate() {
  local name
  local -a fsync_targets=("$ROLLBACK_DIR/prestate.tsv")
  for name in "${PAYLOAD_FILES[@]}"; do
    if [ "${FILE_EXISTED[$name]:-}" = "1" ]; then
      fsync_targets+=("$ROLLBACK_DIR/$name")
    fi
  done
  fsync_paths_and_parents "${fsync_targets[@]}"
}

write_receipt() {
  local receipt_status="$1"
  local rollback_attempted="$2"
  local rollback_performed="$3"
  local rollback_verified="$4"
  local zero_residue_verified="$5"
  install -d -m 700 "$STATUS_ROOT" || return 1
  [ -d "$STATUS_ROOT" ] && [ ! -L "$STATUS_ROOT" ] || return 1
  [ "$(stat -c '%U:%G:%a' "$STATUS_ROOT")" = "wooo:wooo:700" ] || return 1
  if ! python3 - "$RECEIPT_PATH" "$receipt_status" "$SOURCE_REVISION" "$SOURCE_HEAD" "$RUN_ID" \
    "$rollback_attempted" "$rollback_performed" "$rollback_verified" \
    "$zero_residue_verified" "$VERIFIER_DIGEST" "$FAULT_INJECTION_ENABLED" <<'PY'
import hashlib
import json
import os
import sys
import time
from pathlib import Path

path = Path(sys.argv[1])
status = sys.argv[2]
rollback_attempted = sys.argv[6] == "1"
rollback_performed = sys.argv[7] == "1"
rollback_verified = sys.argv[8] == "1"
zero_residue_verified = sys.argv[9] == "1"
if status not in {"verified", "failed_rolled_back", "rollback_unverified"}:
    raise SystemExit(78)
if status == "verified" and (rollback_attempted or rollback_performed or rollback_verified or not zero_residue_verified):
    raise SystemExit(78)
if status == "failed_rolled_back" and not (
    rollback_attempted and rollback_performed and rollback_verified and zero_residue_verified
):
    raise SystemExit(78)
if status == "rollback_unverified" and (not rollback_attempted or rollback_verified):
    raise SystemExit(78)
document = {
    "schemaVersion": "agent99_host110_backup_runtime_deploy_receipt_v2",
    "status": status,
    "sourceRevision": sys.argv[3],
    "sourceHead": sys.argv[4],
    "runId": sys.argv[5],
    "fileCount": 18,
    "backupScriptFileCount": 17,
    "backupHealthExporterIncluded": True,
    "rollbackAttempted": rollback_attempted,
    "rollbackPerformed": rollback_performed,
    "rollbackVerified": rollback_verified,
    "zeroResidueVerified": zero_residue_verified,
    "verifierSha256": sys.argv[10],
    "executorHost": "192.168.0.110",
    "productionServiceRestarted": False,
    "secretValuesRead": False,
    "writtenAt": int(time.time()),
}
payload = (json.dumps(document, ensure_ascii=True, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8")
expected_digest = hashlib.sha256(payload).hexdigest()
temporary = path.with_name(f".{path.name}.tmp-{os.getpid()}")
fault_enabled = sys.argv[11] == "1"
fault = (
    os.environ.get("HOST110_BACKUP_RUNTIME_TEST_RECEIPT_FAULT", "")
    if status == "verified" and fault_enabled
    else ""
)
if fault not in {"", "write", "link", "fsync", "readback"}:
    raise SystemExit(78)
created_final = False
directory_fd = None
try:
    if os.path.lexists(path) or os.path.lexists(temporary):
        raise FileExistsError(path)
    if fault == "write":
        raise OSError("fault_injected_receipt_write")
    descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(descriptor, "wb") as handle:
        if handle.write(payload) != len(payload):
            raise OSError("receipt_short_write")
        handle.flush()
        if fault == "fsync":
            raise OSError("fault_injected_receipt_fsync")
        os.fsync(handle.fileno())
    os.chmod(temporary, 0o600)
    temporary_readback = temporary.read_bytes()
    if temporary_readback != payload or hashlib.sha256(temporary_readback).hexdigest() != expected_digest:
        raise OSError("receipt_temporary_readback_failed")
    if fault == "link":
        raise OSError("fault_injected_receipt_link")
    os.link(temporary, path)
    created_final = True
    directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
    directory_fd = os.open(path.parent, directory_flags)
    os.fsync(directory_fd)
    if fault == "readback":
        raise OSError("fault_injected_receipt_readback")
    readback = path.read_bytes()
    if readback != payload or hashlib.sha256(readback).hexdigest() != expected_digest:
        raise OSError("receipt_final_readback_failed")
    if json.loads(readback.decode("utf-8")) != document:
        raise OSError("receipt_document_readback_failed")
    temporary.unlink()
    os.fsync(directory_fd)
except BaseException:
    try:
        if created_final:
            path.unlink(missing_ok=True)
        temporary.unlink(missing_ok=True)
        if directory_fd is None and path.parent.is_dir():
            directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
            directory_fd = os.open(path.parent, directory_flags)
        if directory_fd is not None:
            os.fsync(directory_fd)
    finally:
        raise
finally:
    if directory_fd is not None:
        os.close(directory_fd)
PY
  then
    return 1
  fi
  [ -f "$RECEIPT_PATH" ] && [ ! -L "$RECEIPT_PATH" ] || return 1
  return 0
}

record_prestate() {
  local name dest_path digest metadata
  : > "$ROLLBACK_DIR/prestate.tsv" || return 74
  chmod 600 "$ROLLBACK_DIR/prestate.tsv" || return 74
  for name in "${PAYLOAD_FILES[@]}"; do
    dest_path="$(destination_path "$name")"
    if [ -f "$dest_path" ] && [ ! -L "$dest_path" ]; then
      [ "$(stat -c '%U:%G' "$dest_path")" = "wooo:wooo" ] || return 71
      FILE_EXISTED[$name]=1
      digest="$(working_digest "$dest_path")" || return 72
      metadata="$(stat -c '%u:%g:%a' "$dest_path")" || return 72
      PREVIOUS_DIGEST[$name]="$digest"
      PREVIOUS_METADATA[$name]="$metadata"
      printf '%s\t1\t%s\t%s\n' "$name" "$digest" "$metadata" >> "$ROLLBACK_DIR/prestate.tsv" || return 72
      cp -p -- "$dest_path" "$ROLLBACK_DIR/$name" || return 72
      [ "$(working_digest "$ROLLBACK_DIR/$name")" = "$digest" ] || return 72
    elif [ ! -e "$dest_path" ]; then
      FILE_EXISTED[$name]=0
      PREVIOUS_DIGEST[$name]="-"
      PREVIOUS_METADATA[$name]="-"
      printf '%s\t0\t-\t-\n' "$name" >> "$ROLLBACK_DIR/prestate.tsv" || return 72
    else
      return 73
    fi
  done
  fsync_rollback_prestate || return 75
}

register_run_owned_temp() {
  RUN_OWNED_TEMP_PATHS["$1"]=1
}

verify_run_owned_temp_absence() {
  local temporary
  for temporary in "${!RUN_OWNED_TEMP_PATHS[@]}"; do
    [ ! -e "$temporary" ] && [ ! -L "$temporary" ] || return 1
  done
}

cleanup_run_owned_temps() {
  local temporary failed=0 preserved=0
  local fault=""
  if [ "$FAULT_INJECTION_ENABLED" -eq 1 ]; then
    fault="${HOST110_BACKUP_RUNTIME_TEST_ROLLBACK_FAULT:-}"
  fi
  for temporary in "${!RUN_OWNED_TEMP_PATHS[@]}"; do
    if [ "$fault" = "preserve_first_temp" ] \
      && [ "$preserved" -eq 0 ] \
      && { [ -e "$temporary" ] || [ -L "$temporary" ]; }; then
      preserved=1
      failed=1
      continue
    fi
    rm -f -- "$temporary" || failed=1
  done
  verify_run_owned_temp_absence || failed=1
  [ "$failed" -eq 0 ]
}

rollback_transaction() {
  local name dest_path destination_parent backup_path temporary failed=0
  ROLLBACK_ATTEMPTED=1
  ROLLBACK_PERFORMED=0
  ROLLBACK_VERIFIED=0
  RUN_TEMP_RESIDUE_VERIFIED=0
  set +e
  cleanup_run_owned_temps || failed=1
  for name in "${PAYLOAD_FILES[@]}"; do
    [ -n "${FILE_EXISTED[$name]+x}" ] || { failed=1; continue; }
    dest_path="$(destination_path "$name")"
    if [ "${FILE_EXISTED[$name]}" = "1" ]; then
      backup_path="$ROLLBACK_DIR/$name"
      destination_parent="$(dirname "$dest_path")"
      temporary="$destination_parent/.${name}.agent99-rollback-${RUN_ID}"
      register_run_owned_temp "$temporary"
      if cp -p -- "$backup_path" "$temporary" && mv -f -- "$temporary" "$dest_path"; then
        ROLLBACK_PERFORMED=1
      else
        failed=1
      fi
    else
      if rm -f -- "$dest_path"; then
        ROLLBACK_PERFORMED=1
      else
        failed=1
      fi
    fi
  done
  cleanup_run_owned_temps || failed=1
  fsync_destination_state || failed=1
  for name in "${PAYLOAD_FILES[@]}"; do
    dest_path="$(destination_path "$name")"
    if [ "${FILE_EXISTED[$name]:-}" = "1" ]; then
      [ -f "$dest_path" ] && [ ! -L "$dest_path" ] \
        && [ "$(working_digest "$dest_path" 2>/dev/null)" = "${PREVIOUS_DIGEST[$name]:-invalid}" ] \
        && [ "$(stat -c '%u:%g:%a' "$dest_path" 2>/dev/null)" = "${PREVIOUS_METADATA[$name]:-invalid}" ] \
        || failed=1
    elif [ "${FILE_EXISTED[$name]:-}" = "0" ]; then
      [ ! -e "$dest_path" ] || failed=1
    else
      failed=1
    fi
  done
  if verify_run_owned_temp_absence; then
    RUN_TEMP_RESIDUE_VERIFIED=1
  else
    failed=1
    RUN_TEMP_RESIDUE_VERIFIED=0
  fi
  set -e
  if [ "$failed" -eq 0 ] && [ "$ROLLBACK_PERFORMED" -eq 1 ] && [ "$RUN_TEMP_RESIDUE_VERIFIED" -eq 1 ]; then
    ROLLBACK_VERIFIED=1
    return 0
  fi
  ROLLBACK_VERIFIED=0
  return 1
}

cleanup() {
  local exit_status=$?
  local rollback_status="rollback_unverified"
  local failure_receipt_written=0
  trap - EXIT HUP INT TERM
  if [ "$APPLY_STARTED" -eq 1 ] && [ "$APPLY_COMPLETE" -ne 1 ]; then
    if rollback_transaction; then
      rollback_status="failed_rolled_back"
    fi
    if write_receipt "$rollback_status" "$ROLLBACK_ATTEMPTED" "$ROLLBACK_PERFORMED" \
      "$ROLLBACK_VERIFIED" "$RUN_TEMP_RESIDUE_VERIFIED"; then
      failure_receipt_written=1
    fi
  fi
  if [ -n "$STAGE_DIR" ]; then
    rm -rf -- "$STAGE_DIR" || true
  fi
  if [ "$ROLLBACK_VERIFIED" -eq 1 ] && [ "$failure_receipt_written" -eq 1 ]; then
    if [ -n "$ROLLBACK_DIR" ]; then
      rm -rf -- "$ROLLBACK_DIR" || true
    fi
  fi
  exit "$exit_status"
}

validate_source || fail "source_validation_failed"

if [ "$MODE" = "check" ]; then
  printf '{"schemaVersion":"agent99_host110_backup_runtime_executor_v1","mode":"check","ok":true,"sourceRevision":"%s","sourceHead":"%s","fileCount":18,"backupScriptFileCount":17,"backupHealthExporterIncluded":true,"remoteWritePerformed":false}\n' "$SOURCE_REVISION" "$SOURCE_HEAD"
  exit 0
fi

if [ "$MODE" = "verify" ]; then
  verify_destination || fail "destination_verification_failed"
  printf '{"schemaVersion":"agent99_host110_backup_runtime_executor_v1","mode":"verify","ok":true,"sourceRevision":"%s","sourceHead":"%s","fileCount":18,"backupScriptFileCount":17,"backupHealthExporterIncluded":true,"remoteWritePerformed":false}\n' "$SOURCE_REVISION" "$SOURCE_HEAD"
  exit 0
fi

[ ! -e "$RECEIPT_PATH" ] && [ ! -L "$RECEIPT_PATH" ] || fail "run_identity_receipt_exists"
STAGE_DIR="$STAGE_ROOT/$RUN_ID"
ROLLBACK_DIR="$ROLLBACK_ROOT/$RUN_ID"
[ ! -e "$STAGE_DIR" ] || fail "run_identity_stage_exists"
[ ! -e "$ROLLBACK_DIR" ] || fail "run_identity_rollback_exists"

[ ! -L "$LOCK_PATH" ] || fail "deployment_lock_symlink_unsafe"
(umask 077; : >> "$LOCK_PATH") || fail "deployment_lock_unavailable"
[ -f "$LOCK_PATH" ] && [ -O "$LOCK_PATH" ] || fail "deployment_lock_owner_invalid"
exec 9>>"$LOCK_PATH"
flock -x -w 30 9 || fail "deployment_lock_timeout"

# Old runtime revisions predate the shared gate. Refuse the first promotion if
# one of them is already active; after common.sh moves first, new starts block.
if pgrep -afu "$EXPECTED_USER" '/backup/scripts/(backup-|check-backup|sync-offsite|verify-offsite|enforce-latest)' \
  | grep -vF "host110-backup-runtime-executor.sh" >/dev/null 2>&1; then
  fail "legacy_backup_runtime_active"
fi

install -d -m 700 "$STAGE_DIR" "$ROLLBACK_DIR"
trap cleanup EXIT
trap 'exit 130' INT
trap 'exit 143' TERM HUP

for name in "${PAYLOAD_FILES[@]}"; do
  install -m 700 "$SOURCE_STAGE/$name" "$STAGE_DIR/$name"
  [ "$(working_digest "$STAGE_DIR/$name")" = "$(source_digest "$name")" ] || fail "stage_identity_failed"
  validate_file "$STAGE_DIR/$name" || fail "stage_validation_failed"
done

record_prestate || fail "rollback_prestate_capture_failed"
APPLY_STARTED=1

# common.sh is promoted first. Every managed mutating backup entrypoint then
# takes the shared side of LOCK_PATH before doing work, while this process holds
# the exclusive side until the independent verifier has completed.
for name in "${PAYLOAD_FILES[@]}"; do
  dest_path="$(destination_path "$name")"
  destination_parent="$(dirname "$dest_path")"
  temporary="$destination_parent/.${name}.agent99-${RUN_ID}"
  register_run_owned_temp "$temporary"
  install -m 755 "$STAGE_DIR/$name" "$temporary"
  if [ "$FAULT_INJECTION_ENABLED" -eq 1 ] \
    && [ "${HOST110_BACKUP_RUNTIME_TEST_APPLY_FAULT:-}" = "after_temp_install" ] \
    && [ "$name" = "${PAYLOAD_FILES[0]}" ]; then
    fail "fault_injected_after_temp_install"
  fi
  mv -f -- "$temporary" "$dest_path"
done

fsync_destination_state || fail "post_apply_durability_failed"
verify_destination || fail "post_apply_verification_failed"
verifier_result="$(BACKUP_RUNTIME_DEPLOY_CONTEXT=1 python3 "$SOURCE_STAGE/verify-host110-backup-runtime.py" \
  --manifest "$SOURCE_STAGE/manifest.json" \
  --destination "$DEST_ROOT" \
  --exporter-destination "$EXPORTER_ROOT/$EXPORTER_FILE" \
  --source-revision "$SOURCE_REVISION" \
  --run-id "$RUN_ID" \
  --expected-verifier-sha256 "$VERIFIER_DIGEST")" || fail "independent_verifier_failed"
python3 - "$verifier_result" "$SOURCE_REVISION" "$RUN_ID" <<'PY' || fail "independent_verifier_contract_failed"
import json
import sys

try:
    row = json.loads(sys.argv[1])
except json.JSONDecodeError:
    raise SystemExit(1)
if not (
    row.get("schemaVersion") == "agent99_host110_backup_runtime_verifier_v1"
    and row.get("ok") is True
    and row.get("sourceRevision") == sys.argv[2]
    and row.get("runId") == sys.argv[3]
    and row.get("fileCount") == 18
    and row.get("backupScriptFileCount") == 17
    and row.get("backupHealthExporterIncluded") is True
    and row.get("remoteWritePerformed") is False
    and row.get("selfIdentityVerified") is True
):
    raise SystemExit(1)
PY

verify_run_owned_temp_absence || fail "post_apply_temp_residue_detected"
RUN_TEMP_RESIDUE_VERIFIED=1
if ! write_receipt "verified" 0 0 0 "$RUN_TEMP_RESIDUE_VERIFIED"; then
  fail "durable_receipt_failed"
fi
receipt_path="$RECEIPT_PATH"
APPLY_COMPLETE=1
trap - EXIT HUP INT TERM
stage_cleanup_verified=0
rollback_prestate_cleanup_verified=0
if rm -rf -- "$STAGE_DIR" && [ ! -e "$STAGE_DIR" ] && [ ! -L "$STAGE_DIR" ]; then
  stage_cleanup_verified=1
fi
if rm -rf -- "$ROLLBACK_DIR" && [ ! -e "$ROLLBACK_DIR" ] && [ ! -L "$ROLLBACK_DIR" ]; then
  rollback_prestate_cleanup_verified=1
fi
STAGE_DIR=""
ROLLBACK_DIR=""
python3 - "$SOURCE_REVISION" "$SOURCE_HEAD" "$receipt_path" "$verifier_result" \
  "$stage_cleanup_verified" "$rollback_prestate_cleanup_verified" <<'PY'
import json
import sys

print(json.dumps({
    "schemaVersion": "agent99_host110_backup_runtime_executor_v1",
    "mode": "apply",
    "ok": True,
    "sourceRevision": sys.argv[1],
    "sourceHead": sys.argv[2],
    "fileCount": 18,
    "backupScriptFileCount": 17,
    "backupHealthExporterIncluded": True,
    "rollbackAttempted": False,
    "rollbackPerformed": False,
    "rollbackVerified": False,
    "zeroResidueVerified": True,
    "stageCleanupVerified": sys.argv[5] == "1",
    "rollbackPrestateCleanupVerified": sys.argv[6] == "1",
    "receipt": sys.argv[3],
    "verifier": json.loads(sys.argv[4]),
}, ensure_ascii=True, sort_keys=True))
PY