#!/usr/bin/env bash
# Dispatch one IPZ Playwright client as a run-owned Kubernetes Job.
#
# This is deliberately a small replacement seam for e2e-remote.  The Job owns
# the server/test stack, while this process owns submission, receipt mapping,
# log streaming, and fail-closed deletion of the run label.
set -Eeuo pipefail
IFS=$'\n\t'
umask 077

SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P)"
MANIFEST_TEMPLATE="$SCRIPT_DIR/../lib/e2e-k8s/job.yaml"
CONTROL_PLANE_IP="${IPZ_E2E_CONTROL_PLANE_IP:-100.101.104.41}"
SSH_USER="${IPZ_E2E_SSH_USER:-user}"
SSH_PORT="${IPZ_E2E_SSH_PORT:-2222}"
SSH_KEY="${IPZ_E2E_SSH_KEY:-${HOME:-/home/user}/.ssh/id_ed25519_buildbox}"
RUNNER_IMAGE="${IPZ_E2E_RUNNER_IMAGE:-}"
SNAPSHOT_DIR="${IPZ_E2E_SNAPSHOT_DIR:-}"
WAIT_SECONDS="${IPZ_E2E_K8S_WAIT_SECONDS:-7200}"
SSH_CONNECT_TIMEOUT="${IPZ_E2E_SSH_CONNECT_TIMEOUT:-10}"

RUN_ID=""
MIRROR_SLUG=""
NODE_REQUEST=""
NODE=""
CLIENT=()
WORK_DIR=""
LOG_PID=""
JOB_SUBMITTED=false
CLEANUP_ATTEMPTED=false
RESULT_SET=false
RESULT_OUTCOME=""
RESULT_STATUS=""
EXPECTED_COMMIT=""
DEBUG_TTL_MINUTES=""
DEBUG_RETENTION_ARMED=false
ACTIVE_DEADLINE_SECONDS=7200
TTL_SECONDS_AFTER_FINISHED=300
ARTIFACT_EXPORTER_SECONDS=0
ARTIFACT_EXPORTER_WAIT_SECONDS=0

usage() {
    local status="${1:-2}"
    cat >&2 <<'USAGE'
usage: e2e-k8s-dispatch [--debug-ttl <minutes>] <run-id> <mirror-slug> <node|auto> -- <client argv...>
       e2e-k8s-dispatch --stub success|server-death|timeout|setup-reject

The normal mode renders the run-owned Job, submits it to the ipz-e2e
namespace through debian3, follows the pod logs, and writes the existing
outcome<TAB>status receipt to $E2E_REMOTE_RECEIPT when that variable is set.
USAGE
    exit "$status"
}

fail() {
    printf 'e2e-k8s-dispatch: %s\n' "$1" >&2
    exit "${2:-2}"
}

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

# A status receipt is intentionally one line and tab-delimited.  The human
# readable stdout lines make the stub acceptance command useful without
# changing the file contract consumed by e2e-remote.
write_receipt() {
    local outcome="$1"
    local status="$2"
    local line="${outcome}"$'\t'"${status}"
    local receipt="${E2E_REMOTE_RECEIPT:-}"

    if [[ -n "$receipt" ]]; then
        local temporary="${receipt}.tmp.$$"
        if ! printf '%s\n' "$line" >"$temporary" || ! mv -f -- "$temporary" "$receipt"; then
            rm -f -- "$temporary" 2>/dev/null || true
            printf 'e2e-k8s-dispatch: unable to write E2E_REMOTE_RECEIPT=%s\n' "$receipt" >&2
            return 1
        fi
    fi

    printf 'receipt: %s\n' "$line"
    printf 'exit-code: %s\n' "$status"
}

set_result() {
    RESULT_OUTCOME="$1"
    RESULT_STATUS="$2"
    RESULT_SET=true
}

node_ip() {
    case "$1" in
        debian1) printf '%s\n' '100.106.253.50' ;;
        debian2) printf '%s\n' '100.79.69.43' ;;
        debian3) printf '%s\n' '100.101.104.41' ;;
        *) return 1 ;;
    esac
}

SSH_BIN="${IPZ_E2E_SSH_BIN:-}"
if [[ -z "$SSH_BIN" ]]; then
    SSH_BIN="$(command -v ssh 2>/dev/null || true)"
fi

ssh_options() {
    printf '%s\n' \
        -p "$SSH_PORT" \
        -i "$SSH_KEY" \
        -o BatchMode=yes \
        -o "ConnectTimeout=$SSH_CONNECT_TIMEOUT" \
        -o StrictHostKeyChecking=accept-new
}

# Every SSH operation sends a script through bash -s.  In particular, no
# kubectl command is assembled as a remote one-liner; this avoids shell-wrapper
# expansion of client arguments and follows the buildbox SSH contract.
remote_control() {
    local operation="$1"
    shift
    local -a options=()
    mapfile -t options < <(ssh_options)
    "$SSH_BIN" "${options[@]}" "$SSH_USER@$CONTROL_PLANE_IP" bash -s -- "$operation" "$@" <<'REMOTE_CONTROL'
set -Eeuo pipefail
IFS=$'\n\t'

operation="${1-}"
shift || true
KUBECTL=(sudo /usr/local/bin/k3s kubectl)

# The runner wrapper and any in-pod server harness use this marker to make the
# client/admission boundary explicit.  A raw container exit code is ambiguous:
# 4 and 5 are valid client statuses as well as the public server-admission
# statuses. Read the marker only after the Playwright container terminates so a
# line in ordinary test output cannot end the wait early; the artifact exporter
# may deliberately keep the overall Pod Running after that point.
read_pod_result() {
    local pod="$1"
    local logs
    local line
    local result=""

    logs="$("${KUBECTL[@]}" logs -n ipz-e2e "$pod" -c playwright --timestamps --tail=-1 2>/dev/null || true)"
    while IFS= read -r line; do
        if [[ "$line" =~ K8S_RESULT[[:space:]]+(client|admission)[[:space:]]+([0-9]+)([[:space:]]|$) ]]; then
            result="${BASH_REMATCH[1]}"$'\t'"${BASH_REMATCH[2]}"
        fi
    done <<< "$logs"

    [[ -n "$result" ]] || return 1
    printf 'K8S_RESULT\t%s\n' "$result"
}

case "$operation" in
    ensure-namespace)
        if ! "${KUBECTL[@]}" get namespace ipz-e2e >/dev/null 2>&1; then
            "${KUBECTL[@]}" create namespace ipz-e2e >/dev/null
        fi
        ;;

    node-ready)
        node="${1:?missing node}"
        ready="$("${KUBECTL[@]}" get node "$node" \
            -o jsonpath='{.status.conditions[?(@.type=="Ready")].status}' 2>/dev/null || true)"
        [[ "$ready" == 'True' ]]
        ;;

    apply)
        manifest_b64="${1:?missing rendered manifest}"
        temporary="$(mktemp -t ipz-e2e-job.XXXXXX)"
        trap 'rm -f -- "$temporary"' EXIT
        printf '%s' "$manifest_b64" | base64 --decode >"$temporary"
        grep -Eq '^kind:[[:space:]]*Job([[:space:]]|$)' "$temporary"
        ! grep -Eq '__[A-Z0-9_]+__' "$temporary"
        "${KUBECTL[@]}" apply -f "$temporary"
        ;;

    stream-logs)
        run_id="${1:?missing run id}"
        selector="app=ipz-e2e,run=$run_id"
        pod=""
        for _ in $(seq 1 120); do
            pod="$("${KUBECTL[@]}" get pods -n ipz-e2e -l "$selector" \
                -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true)"
            if [[ -n "$pod" ]]; then
                break
            fi
            sleep 1
        done
        if [[ -z "$pod" ]]; then
            printf 'e2e-k8s-dispatch: no pod appeared for run %s\n' "$run_id" >&2
            exit 1
        fi
        # The playwright container starts only after the WordPress stack init
        # chain (image pulls + snapshot restore + provisioning) finishes, and
        # `kubectl logs --follow` fails outright while the container is still
        # waiting. Retry until one follow completes, and after the pod turns
        # terminal dump whatever the follow attempts missed so the client's
        # test output is never silently dropped.
        while :; do
            if "${KUBECTL[@]}" logs -n ipz-e2e "$pod" -c playwright --follow --timestamps 2>/dev/null; then
                exit 0
            fi
            phase="$("${KUBECTL[@]}" get pod -n ipz-e2e "$pod" \
                -o jsonpath='{.status.phase}' 2>/dev/null || true)"
            if [[ -z "$phase" || "$phase" == Succeeded || "$phase" == Failed ]]; then
                "${KUBECTL[@]}" logs -n ipz-e2e "$pod" -c playwright --timestamps --tail=-1 2>/dev/null || true
                exit 0
            fi
            sleep 2
        done
        ;;

    wait)
        run_id="${1:?missing run id}"
        deadline_seconds="${2:?missing deadline}"
        node="${3:?missing node}"
        selector="app=ipz-e2e,run=$run_id"
        deadline=$((SECONDS + deadline_seconds))
        missing_job_polls=0

        while ((SECONDS < deadline)); do
            job_json="$("${KUBECTL[@]}" get job -n ipz-e2e -l "$selector" \
                -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true)"
            if [[ -z "$job_json" ]]; then
                missing_job_polls=$((missing_job_polls + 1))
                if ((missing_job_polls >= 3)); then
                    printf 'e2e-k8s-dispatch: nothing ran — the Job for run %s never appeared on %s; check cluster admission logs.\n' "$run_id" "$node" >&2
                    printf 'K8S_RESULT\tadmission\t6\n'
                    exit 0
                fi
                sleep 1
                continue
            fi
            missing_job_polls=0

            succeeded="$("${KUBECTL[@]}" get job -n ipz-e2e -l "$selector" \
                -o jsonpath='{.items[0].status.succeeded}' 2>/dev/null || true)"
            failed="$("${KUBECTL[@]}" get job -n ipz-e2e -l "$selector" \
                -o jsonpath='{.items[0].status.failed}' 2>/dev/null || true)"
            job_reason="$("${KUBECTL[@]}" get job -n ipz-e2e -l "$selector" \
                -o jsonpath='{.items[0].status.conditions[0].reason}' 2>/dev/null || true)"
            pod="$("${KUBECTL[@]}" get pods -n ipz-e2e -l "$selector" \
                -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true)"

            if [[ "$job_reason" == 'DeadlineExceeded' ]]; then
                printf 'K8S_RESULT\tadmission\t5\n'
                exit 0
            fi

            if [[ -n "$pod" ]]; then
                pod_phase="$("${KUBECTL[@]}" get pod -n ipz-e2e "$pod" \
                    -o jsonpath='{.status.phase}' 2>/dev/null || true)"
                scheduled_reason="$("${KUBECTL[@]}" get pod -n ipz-e2e "$pod" \
                    -o jsonpath='{.status.conditions[?(@.type=="PodScheduled")].reason}' 2>/dev/null || true)"
                waiting_reasons="$("${KUBECTL[@]}" get pod -n ipz-e2e "$pod" \
                    -o jsonpath='{range .status.initContainerStatuses[*]}{.state.waiting.reason}{"\\n"}{end}{range .status.containerStatuses[*]}{.state.waiting.reason}{"\\n"}{end}' 2>/dev/null || true)"
                playwright_exit_code="$("${KUBECTL[@]}" get pod -n ipz-e2e "$pod" \
                    -o jsonpath='{.status.containerStatuses[?(@.name=="playwright")].state.terminated.exitCode}' 2>/dev/null || true)"
                init_terminated_codes="$("${KUBECTL[@]}" get pod -n ipz-e2e "$pod" \
                    -o jsonpath='{range .status.initContainerStatuses[*]}{.state.terminated.exitCode}{"\\n"}{end}' 2>/dev/null || true)"
                freshness_exit_code="$("${KUBECTL[@]}" get pod -n ipz-e2e "$pod" \
                    -o jsonpath='{.status.initContainerStatuses[?(@.name=="verify-mirror-revision")].state.terminated.exitCode}' 2>/dev/null || true)"
                terminated_codes="$("${KUBECTL[@]}" get pod -n ipz-e2e "$pod" \
                    -o jsonpath='{range .status.initContainerStatuses[*]}{.state.terminated.exitCode}{"\\n"}{end}{range .status.containerStatuses[*]}{.state.terminated.exitCode}{"\\n"}{end}' 2>/dev/null || true)"

                # The Playwright container's own wrapper is authoritative once
                # that container terminates. The regular artifact exporter keeps
                # the Pod Running during a debug window, so waiting for the Pod's
                # phase to become terminal would turn every retained run into a
                # dispatcher timeout.
                if [[ "$playwright_exit_code" =~ ^[0-9]+$ ]]; then
                    pod_result="$(read_pod_result "$pod" || true)"
                    if [[ "$pod_result" =~ ^K8S_RESULT[[:space:]]+(client|admission)[[:space:]]+([0-9]+)$ ]]; then
                        printf '%s\n' "$pod_result"
                        exit 0
                    fi
                fi

                # A pod that cannot be scheduled is an admission failure.  It
                # has the same public meaning as a server that never listened.
                if [[ "$scheduled_reason" == 'Unschedulable' ]]; then
                    taints="$("${KUBECTL[@]}" get nodes -o jsonpath='{range .items[?(@.metadata.name=="'"$node"'")].spec.taints[*]}{.key}{"\\n"}{end}' 2>/dev/null || true)"
                    if printf '%s\n' "$taints" | grep -qx 'node.kubernetes.io/disk-pressure'; then
                        printf 'e2e-k8s-dispatch: nothing ran — %s is refusing new pods because its disk is nearly full (node.kubernetes.io/disk-pressure). Free space under /home/user/builds on that node.\n' "$node" >&2
                    elif printf '%s\n' "$taints" | grep -qx 'node.kubernetes.io/memory-pressure'; then
                        printf 'e2e-k8s-dispatch: nothing ran — %s is refusing new pods because its memory is exhausted (node.kubernetes.io/memory-pressure). Free memory on that node.\n' "$node" >&2
                    else
                        printf 'e2e-k8s-dispatch: nothing ran — Kubernetes refused to schedule the pod on %s (Unschedulable); inspect the node conditions and taints.\n' "$node" >&2
                    fi
                    printf 'K8S_RESULT\tadmission\t6\n'
                    exit 0
                fi

                # Image pull, hostPath, and init-container setup failures happen
                # before the Playwright client can run.
                case "$waiting_reasons" in
                    *ErrImagePull*|*ImagePullBackOff*|*CreateContainerConfigError*|*CreateContainerError*|*InvalidImageName*)
                        printf 'e2e-k8s-dispatch: nothing ran — pod %s on %s could not create its containers; Kubernetes reported: %s\n' \
                            "$pod" "$node" "$(printf '%s' "$waiting_reasons" | tr '\n' ' ')" >&2
                        printf 'K8S_RESULT\tadmission\t6\n'
                        exit 0
                        ;;
                esac

                if [[ "$playwright_exit_code" =~ ^[0-9]+$ ]]; then
                    # Without the explicit marker, reserved 4/5 statuses are
                    # treated conservatively as server admission outcomes.  The
                    # rendered client wrapper always emits a client marker, so
                    # real client test failures retain their exact status.
                    case "$playwright_exit_code" in
                        4|5)
                            printf 'K8S_RESULT\tadmission\t%s\n' "$playwright_exit_code"
                            ;;
                        *)
                            printf 'K8S_RESULT\tclient\t%s\n' "$playwright_exit_code"
                            ;;
                    esac
                    exit 0
                fi

                if [[ "$pod_phase" == 'Failed' ]]; then
                    # Name the container that killed the run before mapping it
                    # to an opaque admission status; without this, an init or
                    # sidecar failure is indistinguishable from any other
                    # "admission 4" and the pod is deleted moments later.
                    printf 'e2e-k8s-dispatch: pod %s failed before a client marker; container states:\n' "$pod" >&2
                    "${KUBECTL[@]}" get pod -n ipz-e2e "$pod" \
                        -o jsonpath='{range .status.initContainerStatuses[*]}{.name}{"\t"}{.state}{"\n"}{end}{range .status.containerStatuses[*]}{.name}{"\t"}{.state}{"\n"}{end}' >&2 2>/dev/null || true
                    # Preserve an explicit server-side 5 (for example a
                    # readiness timeout) when a stack container reports it;
                    # all other failed-before-client paths are server death.
                    # Exit 42 is reserved for verify-mirror-revision. It proves
                    # that provisioning and the test client never started. Do
                    # not auto-push or repair a stale mirror here: refresh is a
                    # separate operator action.
                    if [[ "$freshness_exit_code" == '42' ]]; then
                        freshness_log="$("${KUBECTL[@]}" logs -n ipz-e2e "$pod" -c verify-mirror-revision --tail=-1 2>/dev/null || true)"
                        freshness_signal="$(printf '%s\n' "$freshness_log" | grep -E '^IPZ_E2E_MIRROR_FRESHNESS expected=(empty|[0-9a-f]{40}) actual=(unreadable|[0-9a-f]{40}) node=(debian1|debian2|debian3)$' | tail -n 1 || true)"
                        if [[ "$freshness_signal" =~ ^IPZ_E2E_MIRROR_FRESHNESS[[:space:]]expected=(empty|[0-9a-f]{40})[[:space:]]actual=(unreadable|[0-9a-f]{40})[[:space:]]node=(debian1|debian2|debian3)$ ]]; then
                            printf 'e2e-k8s-dispatch: nothing ran — mirror on %s is stale or unverifiable (expected %s, actual %s); refresh it before dispatching.\n' \
                                "${BASH_REMATCH[3]}" "${BASH_REMATCH[1]}" "${BASH_REMATCH[2]}" >&2
                            printf 'K8S_RESULT\tadmission\t6\n'
                        else
                            printf 'e2e-k8s-dispatch: mirror freshness init failed without a valid diagnostic; the submitted run is indeterminate.\n' >&2
                            printf 'K8S_RESULT\tadmission\t4\n'
                        fi
                    elif printf '%s\n' "$terminated_codes" | grep -qx '5'; then
                        printf 'K8S_RESULT\tadmission\t5\n'
                    else
                        printf 'K8S_RESULT\tadmission\t4\n'
                    fi
                    exit 0
                fi
            fi

            if [[ "$succeeded" =~ ^[1-9][0-9]*$ ]]; then
                printf 'K8S_RESULT\tclient\t0\n'
                exit 0
            fi
            if [[ "$failed" =~ ^[1-9][0-9]*$ && -z "$pod" ]]; then
                printf 'e2e-k8s-dispatch: nothing ran — the Job failed on %s without creating a pod; inspect cluster admission events.\n' "$node" >&2
                printf 'K8S_RESULT\tadmission\t6\n'
                exit 0
            fi
            sleep 2
        done

        printf 'K8S_RESULT\tadmission\t5\n'
        ;;

    cleanup)
        run_id="${1:?missing run id}"
        if ! "${KUBECTL[@]}" get namespace ipz-e2e >/dev/null 2>&1; then
            exit 0
        fi
        selector="app=ipz-e2e,run=$run_id"
        "${KUBECTL[@]}" delete job -n ipz-e2e -l "$selector" \
            --ignore-not-found=true --wait=true --timeout=30s >/dev/null
        remaining="$("${KUBECTL[@]}" get job -n ipz-e2e -l "$selector" \
            -o name 2>/dev/null || true)"
        [[ -z "$remaining" ]]
        ;;

    *)
        printf 'e2e-k8s-dispatch: unknown remote operation: %s\n' "$operation" >&2
        exit 2
        ;;
esac
REMOTE_CONTROL
}

remote_node_check() {
    local node="$1"
    local operation="$2"
    shift 2
    local ip
    ip="$(node_ip "$node")" || return 1
    local -a options=()
    mapfile -t options < <(ssh_options)
    "$SSH_BIN" "${options[@]}" "$SSH_USER@$ip" bash -s -- "$operation" "$@" <<'REMOTE_NODE'
set -Eeuo pipefail
operation="${1-}"
shift || true
case "$operation" in
    mirror)
        slug="${1:?missing mirror slug}"
        test -d "/home/user/builds/$slug"
        ;;
    image)
        image="${1:?missing image}"
        sudo /usr/local/bin/k3s ctr -n k8s.io images ls -q | grep -Fqx -- "$image"
        ;;
    *)
        printf 'e2e-k8s-dispatch: unknown node operation: %s\n' "$operation" >&2
        exit 2
        ;;
esac
REMOTE_NODE
}

render_manifest() {
    local output="$1"
    shift
    python3 - "$MANIFEST_TEMPLATE" "$output" "$RUN_ID" "$MIRROR_SLUG" "$NODE" "$RUNNER_IMAGE" "$SNAPSHOT_DIR" "$EXPECTED_COMMIT" "$ACTIVE_DEADLINE_SECONDS" "$TTL_SECONDS_AFTER_FINISHED" "$ARTIFACT_EXPORTER_SECONDS" "$ARTIFACT_EXPORTER_WAIT_SECONDS" "${CLIENT[@]}" <<'PY'
from __future__ import annotations

import json
import re
import sys
from pathlib import Path

template_path, output_path, run_id, mirror_slug, node, image, snapshot_dir, expected_commit, active_deadline, finished_ttl, exporter_seconds, exporter_wait_seconds, *client = sys.argv[1:]
text = Path(template_path).read_text(encoding="utf-8")
for token, value in {
    "__RUN_ID__": run_id,
    "__MIRROR_SLUG__": mirror_slug,
    "__NODE__": node,
    "__IMAGE__": image,
    "__SNAPSHOT_DIR__": snapshot_dir,
    "__EXPECTED_COMMIT__": expected_commit,
    "__ACTIVE_DEADLINE_SECONDS__": active_deadline,
    "__TTL_SECONDS_AFTER_FINISHED__": finished_ttl,
    "__ARTIFACT_EXPORTER_SECONDS__": exporter_seconds,
    "__ARTIFACT_EXPORTER_WAIT_SECONDS__": exporter_wait_seconds,
}.items():
    text = text.replace(token, value)

# The shared Job template intentionally leaves the client command open.  Run it
# through a tiny argv-preserving wrapper that emits the explicit client outcome
# marker.  This is what lets wait() distinguish a client exit of 4/5 from the
# same status returned by server bootstrap/readiness admission.
marker = "          workingDir: /workspace/plugins/international-press-zone/tests/e2e\n"
if marker not in text:
    raise SystemExit("playwright container workingDir marker is missing")
if not client:
    raise SystemExit("client argv is empty")
wrapper = (
    "set +e\n"
    "\"$0\" \"$@\"\n"
    "status=$?\n"
    "touch /workspace/.ipz-e2e-client-finished\n"
    "printf 'K8S_RESULT\\tclient\\t%s\\n' \"$status\"\n"
    "exit \"$status\"\n"
)
command = json.dumps(["/bin/sh", "-c"], ensure_ascii=False, separators=(",", ":"))
args = json.dumps([wrapper, *client], ensure_ascii=False, separators=(",", ":"))
text = text.replace(
    marker,
    marker + f"          command: {command}\n          args: {args}\n",
    1,
)
if re.search(r"__[A-Z0-9_]+__", text):
    raise SystemExit("rendered Job still contains an unresolved placeholder")
Path(output_path).write_text(text, encoding="utf-8")
PY
}

cleanup_job() {
    if [[ "$JOB_SUBMITTED" != true || "$CLEANUP_ATTEMPTED" == true || -z "$RUN_ID" ]]; then
        return 0
    fi
    # Debug retention is armed only after apply returns success. An ambiguous
    # apply failure must still clean by run label before it can truthfully remain
    # a setup rejection; otherwise the Job may have started despite status 6.
    # Armed retention is finite in both the exporter and Job controller, with
    # intentionally no unbounded retention path.
    if [[ -n "$DEBUG_TTL_MINUTES" && "$DEBUG_RETENTION_ARMED" == true ]]; then
        return 0
    fi
    CLEANUP_ATTEMPTED=true
    if ! remote_control cleanup "$RUN_ID"; then
        printf 'e2e-k8s-dispatch: fail-closed Job cleanup failed for run %s\n' "$RUN_ID" >&2
        return 1
    fi
}

on_exit() {
    local original_status=$?
    local cleanup_status=0
    local receipt_status
    trap - EXIT INT TERM
    set +e

    if [[ -n "$LOG_PID" ]]; then
        if kill -0 "$LOG_PID" 2>/dev/null; then
            kill "$LOG_PID" 2>/dev/null || true
        fi
        wait "$LOG_PID" 2>/dev/null || true
        LOG_PID=""
    fi

    cleanup_job || cleanup_status=$?

    if ((cleanup_status != 0)); then
        # A surviving run-owned Job is a dispatcher failure, not a client or
        # retryable server-admission result.  Override both receipt and exit
        # status so the caller cannot treat teardown failure as a test result.
        RESULT_OUTCOME=admission
        RESULT_STATUS=1
        RESULT_SET=true
        original_status=1
    fi

    if [[ "$RESULT_SET" == true ]]; then
        receipt_status="$RESULT_STATUS"
        write_receipt "$RESULT_OUTCOME" "$RESULT_STATUS" || {
            printf 'e2e-k8s-dispatch: status receipt write failed\n' >&2
            original_status=1
        }
    elif [[ -n "${E2E_REMOTE_RECEIPT:-}" ]]; then
        # Preserve a receipt even for an unexpected local/SSH failure.  This is
        # an admission failure because no client result was observed.
        receipt_status=4
        write_receipt admission "$receipt_status" || true
        if ((original_status == 0)); then
            original_status="$receipt_status"
        fi
    fi

    if [[ -n "$WORK_DIR" ]]; then
        rm -rf -- "$WORK_DIR"
    fi
    exit "$original_status"
}

# Signal exits take the same cleanup path as normal exits.
trap 'exit 130' INT
trap 'exit 143' TERM
trap on_exit EXIT

# Retention is opt-in and finite. Parse it before every operation that could
# contact a node so invalid values are setup rejections with nothing submitted.
if [[ "${1-}" == '--debug-ttl' ]]; then
    if [[ $# -lt 2 ]]; then
        printf 'e2e-k8s-dispatch: debug TTL value <missing> is invalid; accepted range is 5 to 30 minutes.\n' >&2
        set_result admission 6
        exit 6
    fi
    DEBUG_TTL_MINUTES="$2"
    if ! is_decimal "$DEBUG_TTL_MINUTES" || (( DEBUG_TTL_MINUTES < 5 || DEBUG_TTL_MINUTES > 30 )); then
        printf 'e2e-k8s-dispatch: debug TTL value %s is invalid; accepted range is 5 to 30 minutes.\n' "$DEBUG_TTL_MINUTES" >&2
        set_result admission 6
        exit 6
    fi
    shift 2
    ARTIFACT_EXPORTER_SECONDS=$((DEBUG_TTL_MINUTES * 60))
    TTL_SECONDS_AFTER_FINISHED="$ARTIFACT_EXPORTER_SECONDS"
fi

# Stub mode is parsed before any normal-mode validation and never reaches SSH.
if [[ "${1-}" == '--stub' ]]; then
    [[ $# -eq 2 ]] || usage 2
    case "$2" in
        success)
            set_result client 0
            exit 0
            ;;
        server-death)
            set_result admission 4
            exit 4
            ;;
        timeout)
            set_result admission 5
            exit 5
            ;;
        setup-reject)
            printf 'e2e-k8s-dispatch: nothing ran — stub setup was rejected on the simulated node.\n' >&2
            set_result admission 6
            exit 6
            ;;
        *)
            printf 'e2e-k8s-dispatch: unknown stub outcome: %s\n' "$2" >&2
            usage 2
            ;;
    esac
fi

[[ $# -ge 5 ]] || usage 2
RUN_ID="$1"
MIRROR_SLUG="$2"
NODE_REQUEST="$3"
shift 3
[[ "${1-}" == '--' ]] || usage 2
shift
CLIENT=("$@")
[[ ${#CLIENT[@]} -gt 0 ]] || usage 2

[[ "$RUN_ID" =~ ^[a-z0-9]([-a-z0-9]*[a-z0-9])?$ ]] || \
    fail 'run-id must be a lower-case DNS-1123 label'
(( ${#RUN_ID} <= 55 )) || fail 'run-id is too long for the rendered Job name'
[[ "$MIRROR_SLUG" =~ ^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$ ]] || \
    fail 'mirror-slug contains unsafe path characters'
[[ "$MIRROR_SLUG" != *'..'* ]] || fail 'mirror-slug may not contain ..'
[[ "$NODE_REQUEST" == auto || "$NODE_REQUEST" == debian1 || "$NODE_REQUEST" == debian2 || "$NODE_REQUEST" == debian3 ]] || \
    fail 'node must be auto, debian1, debian2, or debian3'

canonical_playwright=(
    env
    NODE_PATH=/opt/ipz-e2e/node_modules
    /opt/ipz-e2e/node_modules/.bin/playwright
)
playwright_prefix_length=0
playwright_prefix_description=""
playwright_executable="${CLIENT[0]}"
if [[ "$playwright_executable" == env ]]; then
    for (( client_index = 1; client_index < ${#CLIENT[@]}; client_index++ )); do
        if [[ "${CLIENT[client_index]}" =~ ^[A-Za-z_][A-Za-z0-9_]*= ]]; then
            continue
        fi
        playwright_executable="${CLIENT[client_index]}"
        break
    done
fi
case "${CLIENT[0]}" in
    pnpm)
        if [[ "${CLIENT[1]-}" == exec && "${CLIENT[2]-}" == playwright ]]; then
            playwright_prefix_length=3
            playwright_prefix_description="pnpm exec playwright"
        elif [[ "${CLIENT[1]-}" == playwright ]]; then
            playwright_prefix_length=2
            playwright_prefix_description="pnpm playwright"
        fi
        ;;
    npx)
        if [[ "${CLIENT[1]-}" == playwright ]]; then
            playwright_prefix_length=2
            playwright_prefix_description="npx playwright"
        elif [[ "${CLIENT[1]-}" == --no-install && "${CLIENT[2]-}" == playwright ]]; then
            playwright_prefix_length=3
            playwright_prefix_description="npx --no-install playwright"
        fi
        ;;
    playwright)
        playwright_prefix_length=1
        playwright_prefix_description="playwright"
        ;;
    ./node_modules/.bin/playwright)
        playwright_prefix_length=1
        playwright_prefix_description="./node_modules/.bin/playwright"
        ;;
esac

if (( playwright_prefix_length > 0 )); then
    printf "e2e-k8s-dispatch: rewrote '%s' to the absolute runner path — pnpm cannot resolve Playwright inside the Job.\n" \
        "$playwright_prefix_description" >&2
    CLIENT=("${canonical_playwright[@]}" "${CLIENT[@]:playwright_prefix_length}")
elif [[ "${CLIENT[0]}" == env \
    && "${CLIENT[1]-}" == NODE_PATH=/opt/ipz-e2e/node_modules \
    && "${CLIENT[2]-}" == /opt/ipz-e2e/node_modules/.bin/playwright ]]; then
    :
elif [[ "${CLIENT[0]}" == /opt/ipz-e2e/node_modules/.bin/playwright ]]; then
    :
elif [[ "${playwright_executable##*/}" == playwright ]]; then
    printf "e2e-k8s-dispatch: unsupported Playwright executable '%s'; use 'env NODE_PATH=/opt/ipz-e2e/node_modules /opt/ipz-e2e/node_modules/.bin/playwright'.\n" \
        "$playwright_executable" >&2
    set_result admission 6
    exit 6
fi

[[ -r "$MANIFEST_TEMPLATE" ]] || fail "Job template is not readable: $MANIFEST_TEMPLATE"
# Resolve the invoking checkout before any remote operation. A missing or
# unreadable HEAD means there is no truthful revision to compare in the pod.
if ! project_root="$(git -C "$PWD" rev-parse --show-toplevel 2>/dev/null)" \
    || ! EXPECTED_COMMIT="$(git -C "$project_root" rev-parse --verify HEAD 2>/dev/null)" \
    || [[ ! "$EXPECTED_COMMIT" =~ ^[0-9a-f]{40}$ ]]; then
    printf 'e2e-k8s-dispatch: nothing ran — the project root is not a git repository with a resolvable HEAD.\n' >&2
    set_result admission 6
    exit 6
fi
if [[ -z "$RUNNER_IMAGE" ]]; then
    # e2e-runner-image-sync publishes version-pinned tags only (no :latest),
    # so an unset IPZ_E2E_RUNNER_IMAGE must be derived from the invoking
    # checkout's Playwright lockfile with the same exact-version validation.
    [[ -r "$PWD/package.json" && -r "$PWD/package-lock.json" ]] || \
        fail 'set IPZ_E2E_RUNNER_IMAGE or run from the plugin tests/e2e directory (package.json + package-lock.json needed to derive the runner image tag)'
    command -v python3 >/dev/null 2>&1 || fail 'python3 is required to derive the runner image tag'
    playwright_version="$(python3 - "$PWD/package.json" "$PWD/package-lock.json" <<'PY'
import json
import re
import sys

package_path, lock_path = sys.argv[1:]
with open(package_path, encoding="utf-8") as stream:
    package = json.load(stream)
with open(lock_path, encoding="utf-8") as stream:
    lock = json.load(stream)

version = package.get("devDependencies", {}).get("@playwright/test")
if not isinstance(version, str) or not re.fullmatch(r"\d+\.\d+\.\d+", version):
    raise SystemExit("@playwright/test must be an exact semver in package.json")

packages = lock.get("packages", {})
for package_name in ("@playwright/test", "playwright", "playwright-core"):
    lock_entry = packages.get(f"node_modules/{package_name}")
    if not isinstance(lock_entry, dict) or lock_entry.get("version") != version:
        actual = lock_entry.get("version") if isinstance(lock_entry, dict) else None
        raise SystemExit(
            f"{package_name} is {actual!r} in package-lock.json, expected {version!r}"
        )

print(version)
PY
)" || fail 'unable to derive the Playwright version for the runner image tag'
    RUNNER_IMAGE="localhost/ipz-e2e-runner:${playwright_version}"
fi
if [[ -z "$SNAPSHOT_DIR" ]]; then
    # The Job mounts the node-local WordPress snapshot for the exact contract
    # the invoking checkout declares. Derive the fingerprinted directory from
    # that checkout's contract script so a stale or foreign snapshot can never
    # be mounted silently.
    [[ -r "$PWD/e2e-snapshot-contract.sh" ]] || \
        fail 'set IPZ_E2E_SNAPSHOT_DIR or run from the plugin tests/e2e directory (e2e-snapshot-contract.sh needed to derive the snapshot directory)'
    SNAPSHOT_DIR="$(cd "$PWD" && HOME=/home/user bash -c '. ./e2e-snapshot-contract.sh && ipz_snapshot_dir')" || \
        fail 'unable to derive the snapshot directory from e2e-snapshot-contract.sh'
fi
[[ "$SNAPSHOT_DIR" == /* && "$SNAPSHOT_DIR" != *[[:space:]]* && "$SNAPSHOT_DIR" != *$'\n'* ]] || \
    fail 'snapshot directory must be an absolute whitespace-free path'
[[ "$RUNNER_IMAGE" != *[[:space:]]* && "$RUNNER_IMAGE" != *$'\n'* ]] || fail 'IPZ_E2E_RUNNER_IMAGE contains whitespace'
is_decimal "$SSH_PORT" || fail 'IPZ_E2E_SSH_PORT must be numeric'
is_decimal "$SSH_CONNECT_TIMEOUT" || fail 'IPZ_E2E_SSH_CONNECT_TIMEOUT must be numeric'
is_decimal "$WAIT_SECONDS" || fail 'IPZ_E2E_K8S_WAIT_SECONDS must be numeric'
(( WAIT_SECONDS > 0 )) || fail 'IPZ_E2E_K8S_WAIT_SECONDS must be greater than zero'
# Give the client wrapper two minutes after its own dispatcher timeout to write
# the marker. This fixed grace never scales with the optional debug retention.
ARTIFACT_EXPORTER_WAIT_SECONDS=$((WAIT_SECONDS + 120))
# Leave one further minute for the exporter to exit after its marker deadline (or
# its post-marker retention). The Job deadline must never kill this sidecar first.
ACTIVE_DEADLINE_SECONDS=$((ARTIFACT_EXPORTER_WAIT_SECONDS + ARTIFACT_EXPORTER_SECONDS + 60))
[[ -n "$SSH_BIN" && -x "$SSH_BIN" ]] || fail 'ssh is not available'
command -v base64 >/dev/null 2>&1 || fail 'base64 is required'
command -v python3 >/dev/null 2>&1 || fail 'python3 is required to render the Job'

WORK_DIR="$(mktemp -d "${TMPDIR:-/tmp}/ipz-e2e-k8s.XXXXXX")"

if [[ "$NODE_REQUEST" == auto ]]; then
    NODE=""
    auto_order="${IPZ_E2E_NODE_ORDER:-debian1,debian2,debian3}"
    IFS=',' read -r -a auto_nodes <<< "$auto_order"
    for candidate in "${auto_nodes[@]}"; do
        case "$candidate" in
            debian1|debian2|debian3) ;;
            *) continue ;;
        esac
        if remote_control node-ready "$candidate" >/dev/null 2>&1 \
            && remote_node_check "$candidate" mirror "$MIRROR_SLUG" >/dev/null 2>&1 \
            && remote_node_check "$candidate" image "$RUNNER_IMAGE" >/dev/null 2>&1; then
            NODE="$candidate"
            break
        fi
    done
    # Auto mode exhausted every eligible buildbox.  Preserve the harness's
    # public no-usable-host status instead of disguising it as a server death.
    [[ -n "$NODE" ]] || {
        set_result admission 97
        exit 97
    }
else
    NODE="$NODE_REQUEST"
    if ! remote_control node-ready "$NODE" >/dev/null 2>&1; then
        printf 'e2e-k8s-dispatch: nothing ran — requested node %s is not ready; restore that node before dispatching.\n' "$NODE" >&2
        set_result admission 6
        exit 6
    fi
    if ! remote_node_check "$NODE" mirror "$MIRROR_SLUG" >/dev/null 2>&1; then
        printf 'e2e-k8s-dispatch: nothing ran — %s does not have the source mirror %s; push it before dispatching.\n' "$NODE" "$MIRROR_SLUG" >&2
        set_result admission 6
        exit 6
    fi
    if ! remote_node_check "$NODE" image "$RUNNER_IMAGE" >/dev/null 2>&1; then
        printf 'e2e-k8s-dispatch: nothing ran — %s does not have runner image %s imported; import it before dispatching.\n' "$NODE" "$RUNNER_IMAGE" >&2
        set_result admission 6
        exit 6
    fi
fi

rendered_manifest="$WORK_DIR/job.yaml"
if ! render_manifest "$rendered_manifest"; then
    printf 'e2e-k8s-dispatch: nothing ran — the Job manifest for %s on %s could not be rendered; fix the template or client arguments.\n' "$RUN_ID" "$NODE" >&2
    set_result admission 6
    exit 6
fi
manifest_b64="$(base64 <"$rendered_manifest" | tr -d '\n')"

# Namespace creation is not run ownership, so it stays outside the cleanup
# claim.  Ownership starts at the apply ATTEMPT, not at its success: kubectl
# can create the Job and still exit nonzero (connection cut after submission),
# or a signal can land mid-apply.  cleanup_job deletes by this run's label
# with --ignore-not-found, so claiming before apply is idempotent — either
# our Job exists and is removed, or nothing matches and it is a no-op.
if ! remote_control ensure-namespace; then
    printf 'e2e-k8s-dispatch: nothing ran — Kubernetes rejected namespace setup before dispatching to %s.\n' "$NODE" >&2
    set_result admission 6
    exit 6
fi
JOB_SUBMITTED=true
if ! remote_control apply "$manifest_b64"; then
    printf 'e2e-k8s-dispatch: nothing ran — Kubernetes rejected the Job manifest for %s on %s.\n' "$RUN_ID" "$NODE" >&2
    set_result admission 6
    exit 6
fi
# Only a confirmed successful apply is intentionally retainable. The EXIT trap
# continues to delete by run label for every ambiguous/nonzero apply outcome.
DEBUG_RETENTION_ARMED=true

# Logs are followed independently so the wait poller can continue to observe
# the Job's terminal state.
remote_control stream-logs "$RUN_ID" &
LOG_PID=$!

wait_output=""
if ! wait_output="$(remote_control wait "$RUN_ID" "$WAIT_SECONDS" "$NODE")"; then
    printf '%s\n' "$wait_output" >&2
    printf 'e2e-k8s-dispatch: Job %s was submitted to %s but the dispatcher lost contact with Kubernetes before it could read the result; the run may or may not have completed. Restore access and dispatch again.\n' "$RUN_ID" "$NODE" >&2
    set_result admission 4
    exit 4
fi
printf '%s\n' "$wait_output"

result_outcome=""
result_status=""
while IFS=$'\t' read -r marker outcome status; do
    if [[ "$marker" == K8S_RESULT ]]; then
        result_outcome="$outcome"
        result_status="$status"
    fi
done <<< "$wait_output"

if [[ "$result_outcome" != client && "$result_outcome" != admission ]] || \
   ! is_decimal "$result_status"; then
    printf 'e2e-k8s-dispatch: Job %s on %s returned a result the dispatcher could not parse; treat this run as indeterminate and inspect the output above.\n' "$RUN_ID" "$NODE" >&2
    set_result admission 4
    exit 4
fi

case "$result_outcome" in
    client)
        (( result_status <= 255 )) || {
            printf 'e2e-k8s-dispatch: Job %s on %s reported an impossible client exit status %s; treat this run as indeterminate.\n' "$RUN_ID" "$NODE" "$result_status" >&2
            set_result admission 4
            exit 4
        }
        set_result client "$result_status"
        exit "$result_status"
        ;;
    admission)
        case "$result_status" in
            3|4|5|6|97)
                set_result admission "$result_status"
                exit "$result_status"
                ;;
            *)
                printf 'e2e-k8s-dispatch: Job %s on %s reported an unrecognised admission status %s; treat this run as indeterminate.\n' "$RUN_ID" "$NODE" "$result_status" >&2
                set_result admission 4
                exit 4
                ;;
        esac
        ;;
esac