#!/bin/bash
# test-commit-durability.sh — bug-661 reproduction: committed jobs lost across failover.
#
# Spec: ha-commit-durability (.kiro/specs/ha-commit-durability/)
# Reproduces the manual /kiro-discovery session of 2026-06-17 where a 3-node
# controller cluster, driven by xce_test under rolling primary kills, PERMANENTLY
# lost jobs that had already returned "jobsubmit ok" (committed) — joblist dropped
# from 10 jobs to 3.
#
# Root cause (bug-661): "committed" != "durable". WAL fsync is asynchronous
# (persist.c fsync_worker, interval wal_sync_ms); the standby ACKs a WAL entry
# before fdatasync, and commit.c advances committed_lsn on majority *ACK* (page
# cache only). A SIGKILL during the async window drops the un-fsync'd tail; on
# restart persist_recover hits a CRC break and replays ~0 entries, yet the hollow
# node can still win election and overwrite the good timeline.
#
# THE INVARIANT UNDER TEST (Requirement 1, durable commit):
#   Any job that returned "jobsubmit ok" MUST remain visible in joblist for the
#   remainder of the run, across any sequence of single-node SIGKILL + restart.
#   The cluster never loses a committed job (n_failures tolerated = floor(N/2) = 1).
#
# ── soak mode: A SOAK / RACE REPRODUCTION, NOT A DETERMINISTIC ONE ─────────────
#   bug-661 is a commit-durability RACE: wal_sync_ms defaults to 10ms
#   (config.h DEFAULT_WAL_SYNC_MS), so the committed-but-not-yet-fsync'd window is
#   ~10ms wide. Loss only occurs when a SIGKILL lands inside that window on enough
#   nodes. This soak fires bursts and kills with no delay across many rounds to hit
#   it; empirically it surfaces loss within a handful of rounds but NOT on every
#   single round.
#
#   Interpret soak results asymmetrically:
#     FAIL ("DATA LOSS")  => bug-661 PROVEN (a committed job vanished). Definitive.
#     PASS (all rounds)   => the race did not fire this run; NOT proof of durability.
#                            Re-run or raise ROUNDS. After the ha-commit-durability
#                            fix (fsync-before-ACK), it MUST pass deterministically
#                            regardless of timing — that is the real acceptance bar.
#
# ── deterministic mode: WHY IT PROVES THE GAP, NOT SIGKILL DATA-LOSS ───────────
#   SCOPE CLARIFICATION (ratify with reviewer). The task asks for a deterministic
#   mode that "reports DATA LOSS without relying on race timing". But on a SINGLE
#   HOST, committed-job DATA LOSS via SIGKILL is PHYSICALLY NON-DETERMINISTIC, for
#   two independent reasons rooted in the code:
#     1. persist_wal_async (persist.c:340) pwrites entry bytes INLINE (persist.c
#        :375,383) — they reach the OS PAGE CACHE immediately; only fdatasync is
#        deferred to fsync_worker. A SIGKILL does NOT evict page-cache pages: the
#        kernel still flushes them to disk, so an un-fdatasync'd pwrite'd entry is
#        STILL recoverable after SIGKILL+restart on one host.
#     2. Each STANDBY fdatasyncs BEFORE sending its WalAck (persist.c:511-515), so
#        every committed job already has a durable copy on the standbys; killing
#        the primary alone never loses it. The incident's loss needed multi-node
#        kills + a hollow node re-leading before catch-up repair — inherently racy
#        (a prior rolling-primary-kill "deterministic" attempt was REJECTED: the
#        same command gave GREEN/RED/RED across 3 runs — it just swapped the ~10ms
#        fsync race for a hollow-re-lead-vs-catch-up race).
#
#   The bug-661 invariant that IS deterministic is the DURABILITY GAP on the
#   PRIMARY (src/controller/CLAUDE.md "Commit Durability" TARGET RULE; Req 1): when
#   the controller replies "jobsubmit ok", the entry must already be fdatasync'd on
#   a fault-tolerant majority INCLUDING THE PRIMARY's own copy. On the unfixed build
#   the primary defers fdatasync to the fsync_worker timer (every --wal-sync-ms ms),
#   so with a LARGE --wal-sync-ms the primary pwrites but performs ZERO fdatasync
#   between append and ack. THAT gap is deterministic and observable WITHOUT any
#   race: launch each controller under `strace -f -tt -e trace=fdatasync,fsync` and
#   assert that NO primary fdatasync/fsync occurs in the wall-clock window between
#   job submit and the "jobsubmit ok" reply. RED on unfixed build = ack-before-
#   primary-durable. After the fsync-before-ACK fix, a primary fdatasync MUST
#   precede the ack -> GREEN. This verdict is identical on every run (5/5 RED on the
#   unfixed build); it does not depend on timing a kill.
#
# Usage:
#   bash src/tools/test-commit-durability.sh [deploy-dir] [rounds]
#   bash src/tools/test-commit-durability.sh --deterministic [deploy-dir]
#
#   deploy-dir : cluster root (default /tmp/xce-ha-deploy). Deployed if absent.
#   rounds     : number of kill+restart failover rounds (default 8) — soak mode only.
#
#   --deterministic : run the DETERMINISTIC durability-gap proof (strace-based,
#       above) instead of the racy soak. Requires `strace` on PATH (ptrace exec-wrap
#       only — does NOT require PTRACE_ATTACH, which is blocked under
#       yama/ptrace_scope=1 in containers). On the CURRENT (unfixed) build it
#       reports RED (no primary fdatasync precedes the ack). After the fix it must
#       go GREEN. See run_deterministic() below.
#
# Requires a current build: run /xce-build controller first.
set -u

# ── mode selection ────────────────────────────────────────────────────────────
# Default = original racy soak (back-compat for existing callers). Pass
# --deterministic as the FIRST arg to select the deterministic reproduction.
MODE="soak"
if [ "${1:-}" = "--deterministic" ] || [ "${1:-}" = "--det" ]; then
    MODE="deterministic"
    shift
fi

WORKTREE=$(cd "$(dirname "$0")/../.." && pwd)
BINARY="$WORKTREE/src/xce.build/controller/xce_controller"
XCE_TEST="$WORKTREE/src/xce.build/controller/xce_test"
DEPLOY_DIR="${1:-/tmp/xce-ha-deploy}"
ROUNDS="${2:-8}"
PEERS="localhost:9991,localhost:9992,localhost:9993"
MASTERS="localhost:9991,localhost:9992,localhost:9993"
LOGDIR="$DEPLOY_DIR/logs"
NODES=(ctrl-1 ctrl-2 ctrl-3)
QUEUE="${QUEUE:-xce}"
JOBUSER="${JOBUSER:-rshen}"

# Jobs submitted per round (before the kill) and after the new primary settles.
JOBS_BEFORE_KILL="${JOBS_BEFORE_KILL:-3}"
JOBS_AFTER_KILL="${JOBS_AFTER_KILL:-1}"

# Deterministic mode: a LARGE fsync cadence so the primary never fdatasyncs a
# committed entry inside the test window. With this the committed-but-not-durable
# window is the entire run, not a ~10ms race — so an strace of the primary shows
# ZERO fdatasync between a job submit and its "jobsubmit ok" ack. Plumbed into the
# controller launch via the generated start.sh ("$*" passthrough) as --wal-sync-ms.
DET_WAL_SYNC_MS="${DET_WAL_SYNC_MS:-30000}"

# When non-empty, start_node appends these args to start.sh (e.g. --wal-sync-ms).
EXTRA_CTRL_ARGS=""

# Where we remember every JobId the cluster acknowledged. Kept in the worktree
# (not /tmp) so it survives sandbox /tmp teardown between background-poll cycles.
ACKED_FILE="$(mktemp "$WORKTREE/.xce-durability-acked.XXXXXX")"

PASS=0
FAIL=0
INCONC=0   # durability could not be assessed (cluster not serving) — NOT data loss

green()  { printf '\033[32m%s\033[0m\n' "$*"; }
red()    { printf '\033[31m%s\033[0m\n' "$*"; }
yellow() { printf '\033[33m%s\033[0m\n' "$*"; }
pass()   { green "  PASS: $*"; PASS=$((PASS+1)); }
fail()   { red   "  FAIL: $*"; FAIL=$((FAIL+1)); }
# A LIVENESS/AVAILABILITY failure is a real fault, but it is NOT a bug-661
# durability violation — keep it a distinct verdict so it is never mislabeled
# "DATA LOSS". inconc() = durability inconclusive (cluster unreachable), reported
# but kept separate from both PASS and DATA-LOSS so one outage does not cascade
# into a pile of false "committed job vanished" lines.
inconc() { yellow "  INCONCLUSIVE: $*"; INCONC=$((INCONC+1)); }

cleanup_exit() { rm -f "$ACKED_FILE"; kill_cluster 2>/dev/null || true; }
trap cleanup_exit EXIT

# ── cluster lifecycle ─────────────────────────────────────────────────────────

deploy_cluster() {
    if [ ! -f "$BINARY" ] || [ ! -f "$XCE_TEST" ]; then
        red "Binary not found: $BINARY / $XCE_TEST"
        red "Run /xce-build controller first."
        exit 2
    fi
    yellow "Deploying 3-node cluster to $DEPLOY_DIR ..."
    bash "$WORKTREE/src/tools/deploy-ha.sh" "$BINARY" "$DEPLOY_DIR" "$PEERS"
}

start_node() {
    # EXTRA_CTRL_ARGS (e.g. --wal-sync-ms <N>) is passed through start.sh's "$*".
    #
    # STRACE_TRACE=1 (deterministic mode) exec-wraps the controller under strace so
    # its fdatasync/fsync syscalls are timestamped into $LOGDIR/<node>.strace.log.
    # We MUST use ptrace exec-wrap (strace launches the child) — PTRACE_ATTACH
    # (strace -p <pid>) is blocked under yama/ptrace_scope=1 in containers. -f
    # follows the fsync_worker pthread; -tt gives HH:MM:SS.uuuuuu wall-clock so the
    # parser can window syscalls to the submit→ack interval. execve on the
    # controller's self-re-exec (ha_convergence) keeps the tracer attached.
    if [ -n "${STRACE_TRACE:-}" ]; then
        VERBOSE=-vvv strace -f -tt -e trace=fdatasync,fsync \
            -o "$LOGDIR/$1.strace.log" \
            "$DEPLOY_DIR/scripts/start.sh" "$1" $EXTRA_CTRL_ARGS >/dev/null 2>&1 &
    else
        VERBOSE=-vvv "$DEPLOY_DIR/scripts/start.sh" "$1" $EXTRA_CTRL_ARGS >/dev/null 2>&1 &
    fi
}

start_cluster() {
    kill_cluster 2>/dev/null || true
    sleep 0.3
    start_node ctrl-1; sleep 0.5
    start_node ctrl-2; sleep 0.5
    start_node ctrl-3; sleep 0.3
}

kill_cluster() {
    # Exact process-name match (-x): never matches this script's own cmdline,
    # which contains the literal "xce_controller" in its grep/pgrep patterns.
    pkill -9 -x xce_controller 2>/dev/null || true
    sleep 0.5
}

node_pid() { pgrep -f "xce_controller.*--myname $1" 2>/dev/null | head -1; }

# SIGKILL a single named node (drops any WAL tail not yet fsync'd by the timer).
kill_node() {
    local pid; pid=$(node_pid "$1")
    [ -n "$pid" ] && kill -9 "$pid" 2>/dev/null || true
    sleep 0.5
}

# Last role line in a node's log is its current role (tolerates log append/rotate).
node_role() {
    grep -aoE 'CTRL_PRIMARY|CTRL_STANDBY|CTRL_ELECTING' "$LOGDIR/$1.log" 2>/dev/null | tail -1
}

# The live node whose most recent role transition is CTRL_PRIMARY.
current_primary() {
    local n
    for n in "${NODES[@]}"; do
        [ -n "$(node_pid "$n")" ] || continue
        [ "$(node_role "$n")" = "CTRL_PRIMARY" ] && { echo "$n"; return 0; }
    done
    return 1
}

# NOTE: serve-readiness ("is the cluster serving yet") is no longer decided by
# polling controller logs. It is now an xce_test exit-code gate via the
# `wait-primary` primitive (see wait_primary_primitive below). The log helpers
# above (node_role / current_primary / node_pid) survive ONLY for node-lifecycle
# decisions the shell still owns: picking which node to SIGKILL and locating a
# node's strace log — never for a pass/fail verdict.

# ── xce_test primitive drivers (piped stdin; exit code IS the verdict) ─────────
#
# Every cluster interaction now goes through an `xce_test` script whose PROCESS
# EXIT CODE is the pass/fail verdict (xce-test-e2e-primitives Req 1). The shell
# never parses joblist/jobsub prose for a verdict anymore — it pipes a script,
# checks `$?`, and (for jobsub) reads the tool's own authoritative captured id
# off an `echo $jobid` line after the clean-ack gate already passed.
#
# ASAN_OPTIONS=detect_leaks=0 on every invocation: the client has a known
# out-of-scope leak (design.md note); the exit-code verdict is only trustworthy
# with a clean exit, so leak detection must be off for the verdict to be valid.

# Run a piped xce_test script; return its exit code (the verdict). The script is
# passed on stdin; readline reads to EOF. xce_test does its own client-side
# master rotation, so a single invocation rides through a failover.
xce_test_script() {
    local script="$1"
    printf '%s' "$script" \
        | ASAN_OPTIONS=detect_leaks=0 timeout 60 "$XCE_TEST"
}

# Serve-readiness gate (Req 4): block until a client op succeeds against the
# configured masters (or time out). Replaces the old log-polling wait_for_primary
# for the "is the cluster serving yet" decision. Returns 0 on success.
# Arg: timeout in ms (default 40000).
wait_primary_primitive() {
    local timeout_ms="${1:-40000}"
    xce_test_script "$(printf 'set masters %s\nclient\nwait-primary %s\n' \
                              "$MASTERS" "$timeout_ms")"
}

# Cluster HEALTH gate (NOT just "a primary serves"): every node process is back
# up (lifecycle via node_pid) AND a client op succeeds (wait-primary). This is
# the precondition for (a) killing the next primary — so we never SIGKILL down to
# a single node and induce a self-inflicted quorum loss the test then misreports
# as bug-661 — and (b) trusting a durability assertion. A rejoined node that has
# not finished catching up is the usual cause of the next round losing quorum, so
# we also settle briefly once all three are up.  Returns 0 iff healthy in time.
# Args: <timeout_s (default 45)>.
wait_cluster_healthy() {
    local timeout_s="${1:-45}" deadline n alive
    deadline=$(( $(date +%s) + timeout_s ))
    while [ "$(date +%s)" -lt "$deadline" ]; do
        alive=0
        for n in "${NODES[@]}"; do [ -n "$(node_pid "$n")" ] && alive=$((alive+1)); done
        if [ "$alive" -eq "${#NODES[@]}" ] \
           && wait_primary_primitive 5000 >/dev/null 2>&1; then
            sleep "${REJOIN_SETTLE_S:-5}"   # let the rejoined node finish catch-up
            return 0
        fi
        sleep 1
    done
    return 1
}

# Submit one job; echo the acked JobId (empty on failure). The ack VERDICT is the
# xce_test EXIT CODE (clean ack carrying a valid JobId => exit 0; bug-786 fix
# guarantees `jobsub` never reports "ok" with a malformed/uninitialized id and
# returns non-zero otherwise). On exit 0 we read the id from the `echo $jobid`
# line — that is the TOOL's authoritative captured id (after clean-ack gating),
# NOT a prose-grep of "jobsubmit ok". No more malformed-id grep.
xce_submit_one() {
    local out rc id
    # `jobsub` sets $jobid only on a clean ack; we then `echo` it on a uniquely
    # marked line so the shell captures EXACTLY the tool's authoritative id
    # (whatever fmt_jobid form it used — non-array N-N-N-N or array N-N-N-N[i]),
    # format-agnostically, instead of re-imposing a brittle id regex. The marker
    # and $jobid are SEPARATE whitespace-delimited tokens: $jobid is only
    # substituted as a WHOLE token (a `$` glued to other text, e.g.
    # `MARK=$jobid`, is NOT expanded), so we keep them apart and echo prints
    # both. If the submit did not cleanly ack, jobsub fails -> session non-zero
    # -> discard.
    out=$(xce_test_script "$(printf 'set masters %s\nclient\njobsub -q %s -u %s -c "sleep 1000"\necho ACKED_JOBID $jobid\n' \
                                    "$MASTERS" "$QUEUE" "$JOBUSER")")
    rc=$?
    [ "$rc" -ne 0 ] && return 1
    # Exit 0 == clean ack. Pull the id off the marked line (second field, last
    # such line wins). The id is whatever echo emitted after the marker token.
    id=$(echo "$out" | sed -n 's/^ACKED_JOBID //p' | tail -1)
    [ -z "$id" ] && return 1
    echo "$id"
}

submit_n() {
    local n=$1 i jid
    for i in $(seq 1 "$n"); do
        jid=$(xce_submit_one)
        if [ -n "$jid" ]; then
            echo "$jid" >> "$ACKED_FILE"
            echo "    submitted (acked): $jid"
        else
            yellow "    submit did not cleanly ack (cluster mid-failover) — not counted"
        fi
        # Tight burst: keep the freshly-committed tail un-fsync'd when the kill
        # lands (the ~10ms wal_sync_ms window). No long inter-submit delay.
    done
}

# Build ONE xce_test script: `expect-job <id>` per acked id, gate the WHOLE
# no-loss assertion on the single exit code (Req 5, Req 8.3). No joblist prose
# membership parsing — `expect-job` fails the session (non-zero) on the first
# missing committed id and reports it. Returns 0 iff all acked jobs survive.
assert_no_loss() {
    local label="$1" total script rc
    total=$(sort -u "$ACKED_FILE" | grep -c . || true)

    if [ "$total" -eq 0 ]; then
        # Nothing acked yet (e.g. baseline before any clean ack) — vacuously OK.
        pass "$label: no committed jobs yet (nothing to lose)"
        return 0
    fi

    # DURABILITY IS ONLY ASSERTABLE AGAINST A SERVING PRIMARY. If the cluster is
    # not serveable, `expect-job <id> not found` cannot tell a genuinely lost job
    # from a merely-unreachable primary — reporting "DATA LOSS / bug-661" there is
    # a FALSE verdict. Probe first; if the cluster is down, the result is
    # INCONCLUSIVE (a liveness problem the caller already flags), NOT data loss.
    if ! wait_primary_primitive "${SERVE_PROBE_MS:-20000}" >/dev/null 2>&1; then
        inconc "$label: cluster not serving — durability NOT assessed (liveness failure, not data loss)"
        return 2
    fi

    # set masters / client first, then one expect-job per unique acked id.
    # Accumulate with explicit $'\n' — command substitution strips trailing
    # newlines, so `script+=$(printf ...\n)` would concatenate lines into one
    # malformed token.
    script="set masters ${MASTERS}"$'\n'"client"$'\n'
    local jid
    while read -r jid; do
        [ -z "$jid" ] && continue
        script+="expect-job ${jid}"$'\n'
    done < <(sort -u "$ACKED_FILE")

    xce_test_script "$script"
    rc=$?
    if [ "$rc" -eq 0 ]; then
        pass "$label: all $total committed jobs present after failover"
        return 0
    fi

    # First check failed — a committed job is missing from a serving primary. To
    # avoid mislabeling a momentary post-failover CATCH-UP LAG (a just-elected
    # primary still applying replicated entries) as a permanent loss, CONFIRM the
    # absence: settle, re-confirm the cluster is healthy, and re-run the check.
    # bug-661's invariant is that an acked job stays visible — a job still absent
    # after a generous settle is a genuine durability violation; one that
    # reappears was transient catch-up, not loss.
    yellow "  $label: a committed job is missing — confirming (could be catch-up lag) ..."
    sleep "${LOSS_CONFIRM_SETTLE_S:-8}"
    wait_cluster_healthy 30 || true
    xce_test_script "$script"
    rc=$?
    if [ "$rc" -eq 0 ]; then
        pass "$label: all $total committed jobs present after a catch-up settle (transient lag, not loss)"
        return 0
    else
        fail "$label: DATA LOSS — a committed job is PERSISTENTLY absent from a SERVING cluster after a confirm settle (bug-661); see expect-job output above"
        return 1
    fi
}

# Assert the cluster currently lists EXACTLY N jobs (Req 5.1/5.2), gated on the
# xce_test exit code. Used by the cold-restart-all recovery phase. Returns 0 iff
# the live count equals N.
#
# Retries up to 3 times (2s settle sleep between) before declaring a mismatch:
# bug-1396's original "extra job" observation and a fresh repro of this exact
# check both trace to the SAME client-side race as bug-1422 (see
# .wolf/buglog.json) — right after a fresh election (e.g. this phase's cold
# restart), libxce's failover retry loop can exhaust its budget without ever
# reaching the new primary and surface a stale E_XCE_STANDBY from whichever
# standby it last hit, which reads here as "xce_joblist: Unknown error 10014"
# / "could not determine job count" rather than a real count. Confirmed via a
# live repro: the aggregate expect-jobs call failed this way while the very
# next per-id assert_no_loss check (16 individual expect-job calls) found
# every job present. A count still wrong after a real settle + 2 retries is a
# genuine mismatch, not this race.
assert_job_count() {
    local label="$1" expected="$2" rc attempt
    for attempt in 1 2 3; do
        xce_test_script "$(printf 'set masters %s\nclient\nexpect-jobs %s\n' \
                                  "$MASTERS" "$expected")"
        rc=$?
        if [ "$rc" -eq 0 ]; then
            pass "$label: cluster lists exactly $expected jobs"
            return 0
        fi
        if [ "$attempt" -lt 3 ]; then
            yellow "  $label attempt $attempt/3: count check failed (likely post-election client retry-storm race, bug-1422 class) — settling 2s and retrying"
            sleep 2
        fi
    done
    # bug-1396: a bare count mismatch isn't enough to diagnose an extra-job
    # anomaly — dump the live joblist and the harness's own ACKED_FILE
    # ledger side by side so the diverging job ID is visible in the log
    # instead of only the count.
    yellow "  $label: job count mismatch after 3 attempts — dumping live joblist vs ACKED_FILE for diagnosis"
    xce_test_script "$(printf 'set masters %s\nclient\njoblist\n' "$MASTERS")"
    echo "  --- ACKED_FILE (tracked committed jobs, deduplicated) ---"
    sort -u "$ACKED_FILE" | sed 's/^/    /'
    fail "$label: job count != $expected after recovery (see joblist/ACKED_FILE dump above)"
    return 1
}

# ── soak mode (original racy reproduction) ────────────────────────────────────

run_soak() {
    echo ""
    yellow "=== bug-661 SOAK: committed-job durability across rolling failover ==="
    echo "    deploy=$DEPLOY_DIR rounds=$ROUNDS queue=$QUEUE (racy ~10ms wal_sync_ms window)"
    echo ""

    bash "$DEPLOY_DIR/scripts/cleanup.sh" >/dev/null 2>&1 || true
    start_cluster

    # Serve-readiness is now an xce_test exit-code gate (Req 4), not log polling:
    # wait-primary blocks until a client op succeeds against the masters.
    if ! wait_primary_primitive 40000; then
        fail "cluster did not become serveable (wait-primary timed out) within 40s"
        exit 1
    fi
    # current_primary reads logs ONLY to pick which node to SIGKILL — node
    # lifecycle, allowed in the shell. The "is it serving" decision above was
    # wait-primary, not log prose.
    green "Initial primary (for victim selection): $(current_primary || echo '?')"

    # Seed a baseline batch so there is committed state before the first kill.
    echo ""
    echo "--- seeding initial committed jobs ---"
    submit_n 4
    assert_no_loss "baseline" || true   # record, continue even if already broken

    # ── rolling failover rounds ───────────────────────────────────────────────
    local round victim vpid
    for round in $(seq 1 "$ROUNDS"); do
        echo ""
        echo "--- round $round/$ROUNDS ---"

        # Only kill into a HEALTHY cluster (all 3 nodes up + serving). Killing the
        # current primary while the previous round's restarted node has not yet
        # rejoined would drop the live set to ONE node (no quorum) — a
        # self-inflicted availability loss the soak must not cause, and must never
        # misreport as durability loss. If the cluster never gets healthy, that is
        # a real LIVENESS failure (recorded once); skip this round's kill+assert
        # rather than cascading false DATA-LOSS lines.
        if ! wait_cluster_healthy 60; then
            fail "round $round: cluster not healthy (3 nodes up + serving) before kill — LIVENESS failure, not data loss"
            continue
        fi

        submit_n "$JOBS_BEFORE_KILL"

        # Pick the victim from log role (node lifecycle); fall back to round-robin.
        victim=$(current_primary || true)
        [ -z "$victim" ] && victim="${NODES[$(( round % 3 ))]}"
        vpid=$(node_pid "$victim")

        if [ -z "$vpid" ]; then
            yellow "  no live victim found; skipping kill this round"
        else
            # SIGKILL via `!`-shell-escape inside an xce_test script (Req 2): the
            # script confirms the cluster is serveable, then escapes to the shell
            # to drop the primary. SIGKILL (not SIGSTOP) drops any WAL tail not
            # yet fsync'd by the timer. Gate on the xce_test exit code.
            echo "  SIGKILL primary $victim (pid $vpid) via !kill immediately after submit ..."
            if ! xce_test_script "$(printf 'set masters %s\nclient\nwait-primary 5000\n!kill -9 %s\n' \
                                           "$MASTERS" "$vpid")"; then
                yellow "  !kill script returned non-zero (kill may still have landed)"
            fi
        fi

        # Let the surviving two elect a new primary and SERVE again (exit-code
        # gate). Post-kill election SLA = ELECT_TIMEOUT_MS (default 60s; a loaded
        # sandbox can run slow — this is a LIVENESS bound, not a durability one).
        if ! wait_primary_primitive "${ELECT_TIMEOUT_MS:-60000}"; then
            fail "round $round: cluster not serveable after killing $victim (>$(( ${ELECT_TIMEOUT_MS:-60000} / 1000 ))s) — LIVENESS failure, not data loss"
            # keep going; assert_no_loss self-gates and reports INCONCLUSIVE, not loss
        else
            echo "  new primary (for next victim): $(current_primary || echo '?')"
        fi
        sleep 2

        # Submit against the new primary while one node is still down.
        submit_n "$JOBS_AFTER_KILL"

        # Restart the killed node via `!`-shell-escape (Req 2) so it rejoins
        # (reproduces hollow-rejoin election). start.sh is the node-lifecycle
        # entrypoint; we escape to it from inside an xce_test script.
        if [ -n "${victim:-}" ]; then
            echo "  restarting $victim (rejoin) via !bash start.sh ..."
            xce_test_script "$(printf '!VERBOSE=-vvv bash %s %s %s >/dev/null 2>&1 &\n' \
                                      "$DEPLOY_DIR/scripts/start.sh" "$victim" "$EXTRA_CTRL_ARGS")" \
                || yellow "  !bash start.sh returned non-zero"
        fi

        # Wait until the cluster is fully HEALTHY again (rejoined node back up +
        # serving + caught up) before asserting durability — so the verdict
        # reflects a real cluster, not a mid-recovery transient. If it never
        # returns to health that is a LIVENESS failure (recorded distinctly);
        # assert_no_loss then self-detects the non-serving state as INCONCLUSIVE
        # rather than crying DATA LOSS.
        if ! wait_cluster_healthy 60; then
            fail "round $round: cluster did not return to healthy (3 up + serving) after $victim rejoined — LIVENESS failure, not data loss"
        fi

        assert_no_loss "round $round" || true
    done

    echo ""
    echo "--- final durability check ---"
    sleep 2
    assert_no_loss "final" || true

    # ── NEW: cold-restart-all + expect-jobs recovery phase (Req 8.3) ──────────
    # A full cold stop + cold start of every node, then assert ALL committed
    # jobs recover. This exercises the durability/recovery path end-to-end (a
    # bug-778-class WAL-replay regression would surface as a count mismatch).
    run_cold_restart_recovery
}

# Full cold restart of the cluster, then assert every acked job survives a
# from-disk recovery. The cold stop/start lifecycle is driven via `!`-shell-
# escape inside xce_test scripts (the task's `!`-driven cold-restart-all path,
# Req 2); the SERVE-readiness gate and the count assertion go through the
# primitives (wait-primary + expect-jobs / expect-job), exit-code-gated.
run_cold_restart_recovery() {
    local committed
    committed=$(sort -u "$ACKED_FILE" | grep -c . || true)

    echo ""
    yellow "=== cold-restart-all recovery: assert $committed committed jobs survive a full restart ==="

    # Cold stop EVERY node via `!`-shell-escape (Req 2): we escape to the shell
    # to kill the whole cluster, proving the `!`-driven cold path.
    echo "  cold stop (!kill all nodes) ..."
    xce_test_script "$(printf '!pkill -9 -x xce_controller; sleep 1\n')" \
        || true   # pkill exits 1 when nothing matched; harmless here.
    sleep 1

    # Cold start every node from disk via `!bash start.sh` (Req 2).
    echo "  cold start (!bash start.sh all nodes) ..."
    local n
    for n in "${NODES[@]}"; do
        xce_test_script "$(printf '!VERBOSE=-vvv bash %s %s %s >/dev/null 2>&1 &\n' \
                                  "$DEPLOY_DIR/scripts/start.sh" "$n" "$EXTRA_CTRL_ARGS")" \
            || yellow "  !bash start.sh $n returned non-zero"
        sleep 0.5
    done

    # SERVE-readiness after the cold restart (exit-code gate, generous timeout:
    # a from-disk recovery + election can take longer than a live failover).
    if ! wait_primary_primitive 60000; then
        fail "cold-restart: cluster not serveable within 60s after full cold restart"
        return 1
    fi
    sleep 3

    # Assert ALL committed jobs recovered after the full cold restart. We check
    # both the exact count (expect-jobs) and per-id membership (expect-job),
    # each gated on the xce_test exit code — no joblist prose parsing.
    assert_job_count "cold-restart" "$committed" || true
    assert_no_loss   "cold-restart" || true
}

# ── deterministic mode (timing-independent DURABILITY-GAP proof) ──────────────
#
# WHAT THIS PROVES (and why it is deterministic — see the header SCOPE
# CLARIFICATION for why single-host SIGKILL data-loss is NOT deterministic):
#   With --wal-sync-ms=$DET_WAL_SYNC_MS (default 30s) the primary's fsync_worker
#   timer never fires inside the test window, so EVERY entry the primary appends
#   stays in the OS page cache, never fdatasync'd (persist_wal_async, the append
#   path, does NOT fsync — only the timer does). Each controller runs under strace
#   capturing fdatasync/fsync with wall-clock timestamps. We submit one job, record
#   the wall-clock instant T0 just before submit and T1 the instant "jobsubmit ok"
#   returns, then assert: ZERO fdatasync/fsync occurred on the PRIMARY in [T0,T1].
#   That is the bug-661 durability gap — the controller acked an entry it had not
#   made durable on its own disk (src/controller/CLAUDE.md "Commit Durability"
#   TARGET RULE / Req 1). The standbys DO fdatasync before WalAck (persist.c:515),
#   so they show syncs in-window; only the primary's missing sync is the defect.
#
#   VERDICT IS RACE-FREE: it depends only on whether the primary issued an
#   fdatasync before the ack returned, which is fixed by the code path (timer-
#   deferred vs sync-before-ack), not by when a kill lands. Same result every run.
#     RED  (unfixed): 0 primary fdatasync/fsync in [T0,T1]  -> ack-before-durable.
#     GREEN (fixed):  >=1 primary fdatasync/fsync in [T0,T1] preceding the ack.
#
# IMPLEMENTATION NOTE: strace ptrace exec-wrap only (start_node launches the
# controller under strace). PTRACE_ATTACH (strace -p) is blocked under
# yama/ptrace_scope=1 in containers, so we never attach to a running pid.

# Convert an strace -tt timestamp "HH:MM:SS.uuuuuu" (or a `date +%H:%M:%S.%N`
# capture) to seconds-of-day as a float. Pure awk; handles the day-wrap-free
# common case (a submit never straddles midnight in this harness).
_hms_to_sod() {
    awk -F'[:.]' '{ printf "%d.%06d\n", ($1*3600+$2*60+$3), substr($4"000000",1,6) }' <<<"$1"
}

# Count PRIMARY fdatasync/fsync syscalls whose -tt timestamp falls in [T0,T1].
# Args: <strace-log> <T0 sod> <T1 sod>. Echoes the count.
_primary_syncs_in_window() {
    local log="$1" t0="$2" t1="$3"
    [ -f "$log" ] || { echo 0; return; }
    # strace -f -tt line shape:  "<pid> HH:MM:SS.uuuuuu fdatasync(<fd>) = 0"
    awk -v t0="$t0" -v t1="$t1" '
        /fdatasync|fsync/ {
            ts=$2                       # HH:MM:SS.uuuuuu
            n=split(ts, a, /[:.]/)
            if (n < 4) next
            sod=a[1]*3600 + a[2]*60 + a[3] + a[4]/1000000.0
            if (sod >= t0 && sod <= t1) c++
        }
        END { print c+0 }' "$log"
}

# One deterministic probe: submit a job, return RED/GREEN by counting primary
# fdatasync/fsync syscalls strictly inside the submit→ack window. Echoes a verdict
# line and sets PASS/FAIL. Returns 0 if RED (gap present), 1 if GREEN.
det_probe_once() {
    local label="$1"
    local primary primary_log t0 t1 t0_sod t1_sod jid syncs

    primary=$(current_primary || true)
    if [ -z "$primary" ]; then
        fail "$label: no primary to probe"
        return 2
    fi
    primary_log="$LOGDIR/$primary.strace.log"
    if [ ! -f "$primary_log" ]; then
        fail "$label: primary strace log missing ($primary_log) — strace exec-wrap failed?"
        return 2
    fi

    # Capture the window with the SAME wall clock strace -tt uses (local time).
    t0=$(date +%H:%M:%S.%N)
    jid=$(xce_submit_one)
    t1=$(date +%H:%M:%S.%N)

    if [ -z "$jid" ]; then
        yellow "  $label: submit returned no JobId (cluster mid-failover?) — retrying once"
        t0=$(date +%H:%M:%S.%N); jid=$(xce_submit_one); t1=$(date +%H:%M:%S.%N)
    fi
    if [ -z "$jid" ]; then
        fail "$label: could not obtain a committed JobId to probe"
        return 2
    fi
    echo "$jid" >> "$ACKED_FILE"

    # Let strace flush the in-window syscalls (its output is line-buffered to file).
    sleep 1
    t0_sod=$(_hms_to_sod "$t0"); t1_sod=$(_hms_to_sod "$t1")
    syncs=$(_primary_syncs_in_window "$primary_log" "$t0_sod" "$t1_sod")

    echo "    primary=$primary  ack=$jid  window=[$t0 .. $t1]  primary fdatasync/fsync in-window=$syncs"
    if [ "$syncs" -eq 0 ]; then
        fail "$label: DURABILITY GAP — primary acked '$jid' with 0 fdatasync in-window (bug-661: ack-before-durable)"
        return 0
    else
        pass "$label: primary fdatasync ($syncs) preceded ack of '$jid' (durable-before-ack)"
        return 1
    fi
}

run_deterministic() {
    if ! command -v strace >/dev/null 2>&1; then
        red "deterministic mode requires 'strace' on PATH (not found) — cannot run."
        red "Re-run in an environment with strace, or use the soak mode."
        exit 2
    fi
    EXTRA_CTRL_ARGS="--wal-sync-ms $DET_WAL_SYNC_MS"
    STRACE_TRACE=1            # start_node will exec-wrap each controller under strace
    local probes="${DET_PROBES:-3}"

    echo ""
    yellow "=== bug-661 DETERMINISTIC: ack-before-durable gap on the PRIMARY (strace) ==="
    echo "    deploy=$DEPLOY_DIR queue=$QUEUE probes=$probes --wal-sync-ms=$DET_WAL_SYNC_MS (no primary fsync in-window)"
    echo "    RED = primary acks a job with 0 fdatasync in the submit->ack window."
    echo ""

    bash "$DEPLOY_DIR/scripts/cleanup.sh" >/dev/null 2>&1 || true
    # Clear stale strace logs so we never read a previous run's syscalls.
    rm -f "$LOGDIR"/*.strace.log 2>/dev/null || true
    start_cluster

    # Serve-readiness via the exit-code primitive (Req 4), not log polling. The
    # log-role current_primary below is used ONLY to locate the strace log of the
    # primary to inspect — node lifecycle, allowed in the shell.
    if ! wait_primary_primitive 40000; then
        fail "cluster did not become serveable (wait-primary timed out) within 40s"
        exit 1
    fi
    # Generous initial settle so the elected primary fully serves jobsub and its
    # boot-time fdatasyncs (segment preallocation) are well before any probe window.
    sleep 6
    green "Initial primary (strace target): $(current_primary)"

    # Repeat the probe several times against the stable primary. Every probe must
    # be RED on the unfixed build — this is the in-run reproducibility evidence
    # (the outer 5x driver gives cross-run evidence).
    local i red_count=0
    for i in $(seq 1 "$probes"); do
        echo ""
        echo "--- det probe $i/$probes ---"
        if det_probe_once "probe $i"; then
            red_count=$((red_count+1))
        fi
        sleep 1
    done

    echo ""
    echo "--- deterministic summary ---"
    echo "    RED probes (durability gap observed): $red_count/$probes"
    # The deterministic mode FAILS (exit non-zero) when the gap is present on the
    # unfixed build (any RED probe). After the fix, all probes go GREEN and FAIL=0.
}

# ── setup + dispatch ──────────────────────────────────────────────────────────

[ -d "$DEPLOY_DIR" ] || deploy_cluster

if [ "$MODE" = "deterministic" ]; then
    run_deterministic
else
    run_soak
fi

# ── final verdict ─────────────────────────────────────────────────────────────

echo ""
echo "========================="
echo "Mode: $MODE"
echo "Results: ${PASS} passed, ${FAIL} failed, ${INCONC} inconclusive"
echo "Acked jobs total: $(sort -u "$ACKED_FILE" | grep -c .)"
if [ "${INCONC:-0}" -gt 0 ]; then
    yellow "NOTE: ${INCONC} durability check(s) were INCONCLUSIVE because the cluster"
    yellow "      was not serving at the time — a LIVENESS/availability problem, NOT a"
    yellow "      bug-661 durability loss. 'FAIL' counts genuine faults only: a DATA LOSS"
    yellow "      is reported solely when a committed job is absent from a SERVING cluster."
fi
echo "========================="
# Exit non-zero on any genuine fault (real DATA LOSS on a serving cluster, or a
# cluster that never returned to health). Inconclusive durability checks do NOT
# by themselves fail the run, but a liveness failure that produced them is itself
# recorded as a FAIL above.
[ "$FAIL" -eq 0 ] && exit 0 || exit 1