diff --git a/etc/launch-kafka-local/compose.connect.yaml b/etc/launch-kafka-local/compose.connect.yaml
new file mode 100644
index 0000000000000..255dd2838a9d6
--- /dev/null
+++ b/etc/launch-kafka-local/compose.connect.yaml
@@ -0,0 +1,50 @@
+# Copyright 2026 Matrix Origin
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+
+services:
+  kafka:
+    image: apache/kafka@sha256:fbc7d7c428e3755cf36518d4976596002477e4c052d1f80b5b9eafd06d0fff2f
+    environment:
+      KAFKA_LOG_DIRS: /var/lib/kafka/data
+      KAFKA_LISTENERS: INTERNAL://0.0.0.0:29092,HOST://0.0.0.0:9092,CONTROLLER://localhost:9093
+      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:29092,HOST://127.0.0.1:${KAFKA_PORT}
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,HOST:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
+    volumes:
+      - kafka-data:/var/lib/kafka/data
+
+  connect:
+    image: ${KAFKA_CONNECT_IMAGE:-quay.io/debezium/connect@sha256:1290a7abb0c1fe957f8d732e43efbea24201d9ba0fa5ba79582f76438c05d0ba}
+    depends_on:
+      kafka:
+        condition: service_healthy
+    ports:
+      - "127.0.0.1:${CONNECT_PORT}:8083"
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+    networks: [kafka-e2e]
+    environment:
+      BOOTSTRAP_SERVERS: kafka:29092
+      GROUP_ID: mo-issue-29848-connect
+      CONFIG_STORAGE_TOPIC: __connect-configs
+      OFFSET_STORAGE_TOPIC: __connect-offsets
+      STATUS_STORAGE_TOPIC: __connect-status
+      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 1000
+      REST_HOST_NAME: 0.0.0.0
+      REST_PORT: 8083
+      ADVERTISED_HOST_NAME: connect
+      ADVERTISED_PORT: 8083
+      KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
+      CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE: "true"
+      VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
+      CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: "true"
+      CONNECT_PLUGIN_PATH: /kafka/connect
+      CONNECT_CONSUMER_AUTO_OFFSET_RESET: earliest
+
+volumes:
+  kafka-data:
diff --git a/optools/kafka_connect_ci.bash b/optools/kafka_connect_ci.bash
new file mode 100755
index 0000000000000..74bccb841a771
--- /dev/null
+++ b/optools/kafka_connect_ci.bash
@@ -0,0 +1,244 @@
+#!/usr/bin/env bash
+# Copyright 2026 Matrix Origin
+# Licensed under the Apache License, Version 2.0.
+
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+PROFILE="${1:-e2e-local}"
+COMPOSE_BASE="$ROOT_DIR/etc/launch-kafka-local/compose.yaml"
+COMPOSE_CONNECT="$ROOT_DIR/etc/launch-kafka-local/compose.connect.yaml"
+REPORT_DIR="${MO_KAFKA_CONNECT_REPORT_DIR:-$ROOT_DIR/test/kafkaconnect/reports/ci_$(date -u +%Y%m%dT%H%M%SZ)}"
+TMP_DIR=""
+MO_PID=""
+PORT_LEASE_PID=""
+PORT_BLOCK_BASE=""
+MO_PORT=""
+STATUS_PORT=""
+KAFKA_PORT=""
+CONNECT_PORT=""
+COMPOSE_PROJECT_NAME=""
+CURRENT_STAGE="preflight"
+SOURCE_SHA=""
+CLEANUP_ERRORS=()
+
+readonly PORT_BLOCK_WIDTH=80
+
+source "$ROOT_DIR/optools/mo_e2e_common.bash"
+
+log() { printf '[kafka-connect-ci] %s\n' "$*"; }
+die() { printf '[kafka-connect-ci] ERROR: %s\n' "$*" >&2; exit 1; }
+require() { command -v "$1" >/dev/null 2>&1 || die "required command not found: $1"; }
+record_cleanup_error() {
+  CLEANUP_ERRORS+=("$1")
+  log "cleanup failed: $1"
+}
+
+e2e_tmp_root() {
+  if [[ "$(uname -s)" == Darwin ]]; then
+    printf '%s\n' /private/tmp
+  else
+    printf '%s\n' /tmp
+  fi
+}
+
+compose() {
+  KAFKA_PORT="$KAFKA_PORT" CONNECT_PORT="$CONNECT_PORT" \
+    docker compose -p "$COMPOSE_PROJECT_NAME" -f "$COMPOSE_BASE" -f "$COMPOSE_CONNECT" "$@"
+}
+
+acquire_port_lease() {
+  local status=0
+  local lease_dir="${MO_MONGODB_PORT_LEASE_DIR:-/tmp/mo-mongodb-port-leases-$(id -u)}"
+  mo_port_block_acquire_lease "$1" "$lease_dir" "${TMPDIR:-/tmp}" || status=$?
+  if (( status == 2 )); then
+    die "could not create or secure Kafka Connect E2E port-block lease"
+  fi
+  return "$status"
+}
+
+allocate_ports() {
+  local requested="${MO_KAFKA_CONNECT_LOG_PORT_BASE:-}" candidate
+  if [[ -n "$requested" ]]; then
+    [[ "$requested" =~ ^[0-9]+$ ]] || die "MO_KAFKA_CONNECT_LOG_PORT_BASE must be an integer"
+    (( requested % PORT_BLOCK_WIDTH == 0 )) || die "MO_KAFKA_CONNECT_LOG_PORT_BASE must align to an ${PORT_BLOCK_WIDTH}-port block"
+    (( requested >= 36000 && requested + PORT_BLOCK_WIDTH <= 65536 )) || die "MO_KAFKA_CONNECT_LOG_PORT_BASE is outside the supported range"
+    PORT_BLOCK_BASE="$requested"
+    acquire_port_lease "$PORT_BLOCK_BASE" || \
+      die "MO_KAFKA_CONNECT_LOG_PORT_BASE is already leased by another local E2E run"
+    mo_port_block_is_available "$PORT_BLOCK_BASE" "$PORT_BLOCK_WIDTH" || {
+      mo_port_block_release_lease
+      die "MO_KAFKA_CONNECT_LOG_PORT_BASE is not available on this host"
+    }
+  else
+    while IFS= read -r candidate; do
+      PORT_BLOCK_BASE="$candidate"
+      if ! acquire_port_lease "$PORT_BLOCK_BASE"; then
+        PORT_BLOCK_BASE=""
+        continue
+      fi
+      if mo_port_block_is_available "$PORT_BLOCK_BASE" "$PORT_BLOCK_WIDTH"; then
+        break
+      fi
+      mo_port_block_release_lease
+      PORT_BLOCK_BASE=""
+    done < <(mo_port_block_candidates "$PORT_BLOCK_WIDTH")
+    [[ -n "$PORT_BLOCK_BASE" ]] || die "could not find an available 80-port range for the Kafka Connect E2E cluster"
+  fi
+
+  MO_PORT=$((PORT_BLOCK_BASE + 72))
+  STATUS_PORT=$((PORT_BLOCK_BASE + 73))
+  KAFKA_PORT=$((PORT_BLOCK_BASE + 74))
+  CONNECT_PORT=$((PORT_BLOCK_BASE + 75))
+}
+
+collect_artifacts() {
+  [[ -n "$TMP_DIR" ]] || return 0
+  mkdir -p "$REPORT_DIR" || return 1
+  if [[ -f "$TMP_DIR/mo-service.log" ]]; then
+    cp "$TMP_DIR/mo-service.log" "$REPORT_DIR/mo-service.log" || return 1
+  fi
+  compose logs --no-color >"$TMP_DIR/connect-compose.raw.log" 2>&1 || return 1
+  python3 - "$TMP_DIR/connect-compose.raw.log" "$REPORT_DIR/connect-compose.log" <<'PY' || return 1
+import re
+import sys
+
+source, destination = sys.argv[1:]
+patterns = [
+    re.compile(r"(?i)(connection\.password\s*[=:]\s*)(\"[^\"]*\"|[^,\s}]+)"),
+    re.compile(r"(?i)(password\s*[=:]\s*)(\"[^\"]*\"|[^,\s}]+)"),
+]
+with open(source, encoding="utf-8", errors="replace") as input_file, open(destination, "w", encoding="utf-8") as output_file:
+    for line in input_file:
+        for pattern in patterns:
+            line = pattern.sub(r"\1<redacted>", line)
+        output_file.write(line)
+PY
+  compose ps --all --format json >"$REPORT_DIR/connect-compose-ps.json" 2>&1 || return 1
+}
+
+cleanup() {
+  local status=$?
+  local service_pid="$MO_PID"
+  set +e
+  set +u
+  if ! mo_e2e_stop_service "$service_pid"; then
+    record_cleanup_error "could not stop the owned MatrixOne process"
+  fi
+  if [[ -n "$service_pid" ]] && kill -0 "$service_pid" >/dev/null 2>&1; then
+    record_cleanup_error "owned MatrixOne process remains after shutdown"
+  fi
+  MO_PID=""
+  if ! collect_artifacts; then
+    record_cleanup_error "could not collect run artifacts"
+  fi
+  if [[ -n "$TMP_DIR" ]]; then
+    if ! compose down --volumes --remove-orphans >/dev/null 2>&1; then
+      record_cleanup_error "docker compose down failed for owned project $COMPOSE_PROJECT_NAME"
+    fi
+    if [[ -d "$TMP_DIR" && "$(basename "$TMP_DIR")" == mo-kafka-connect-e2e.* ]]; then
+      if ! rm -rf -- "$TMP_DIR"; then
+        record_cleanup_error "could not remove owned temporary directory $TMP_DIR"
+      fi
+    else
+      record_cleanup_error "refusing to remove unexpected temporary path: $TMP_DIR"
+    fi
+  fi
+  mo_port_block_release_lease
+  if ((${#CLEANUP_ERRORS[@]} > 0)); then
+    (( status != 0 )) || status=1
+    if [[ -n "$REPORT_DIR" && -f "$REPORT_DIR/report.json" ]]; then
+      python3 - "$REPORT_DIR/report.json" "${CLEANUP_ERRORS[@]}" <<'PY'
+import json
+import sys
+
+path, *cleanup_errors = sys.argv[1:]
+with open(path, encoding="utf-8") as report_file:
+    report = json.load(report_file)
+report["cleanup_status"] = "failed"
+report["cleanup_errors"] = cleanup_errors
+report["status"] = "failed"
+message = "cleanup failed: " + "; ".join(cleanup_errors)
+report["error"] = "; ".join(value for value in (report.get("error"), message) if value)
+with open(path, "w", encoding="utf-8") as report_file:
+    json.dump(report, report_file, indent=2)
+    report_file.write("\n")
+PY
+    fi
+  fi
+  if (( status != 0 )) && [[ -n "$REPORT_DIR" && ! -f "$REPORT_DIR/report.json" ]]; then
+    mkdir -p "$REPORT_DIR"
+    python3 - "$REPORT_DIR/setup-failure.json" "$CURRENT_STAGE" "$SOURCE_SHA" "$status" "${CLEANUP_ERRORS[@]}" <<'PY'
+import json
+import sys
+
+path, stage, sha, status, *cleanup_errors = sys.argv[1:]
+with open(path, "w") as report:
+    json.dump({"status": "failed", "stage": stage, "source_sha": sha, "exit_status": int(status), "cleanup_errors": cleanup_errors}, report, indent=2)
+    report.write("\n")
+PY
+  fi
+  trap - EXIT
+  exit "$status"
+}
+
+run_e2e() {
+  mkdir -p "$REPORT_DIR"
+  rm -f -- "$REPORT_DIR/report.json" "$REPORT_DIR/setup-failure.json"
+  trap cleanup EXIT
+  trap 'exit 130' INT
+  trap 'exit 143' TERM
+
+  require docker
+  require go
+  require python3
+  require make
+  require shasum
+  HARNESS_SHA="$(git -C "$ROOT_DIR" rev-parse HEAD)"
+  SOURCE_SHA="$HARNESS_SHA"
+  local tmp_root
+  tmp_root="$(e2e_tmp_root)"
+  TMP_DIR="$(mktemp -d "${tmp_root%/}/mo-kafka-connect-e2e.XXXXXX")"
+  COMPOSE_PROJECT_NAME="mo-kc-$(basename "$TMP_DIR" | tr '[:upper:].' '[:lower:]-')"
+
+  CURRENT_STAGE="allocate_ports"
+  allocate_ports
+  export MO_PORT KAFKA_PORT CONNECT_PORT
+  log "owned port block $PORT_BLOCK_BASE-$((PORT_BLOCK_BASE + PORT_BLOCK_WIDTH - 1)); MO=$MO_PORT Kafka=$KAFKA_PORT Connect=$CONNECT_PORT"
+
+  CURRENT_STAGE="build_matrixone"
+  local service_bin="$TMP_DIR/mo-service"
+  (cd "$ROOT_DIR" && make BIN_NAME="$service_bin" build-with-prebuilt-native)
+  mo_e2e_generate_launch_config "$ROOT_DIR" "$TMP_DIR" "$PORT_BLOCK_BASE" "$MO_PORT" "$STATUS_PORT"
+
+  if [[ "$(uname -s)" == Darwin ]]; then
+    export DYLD_LIBRARY_PATH="$ROOT_DIR/cgo:$ROOT_DIR/thirdparties/install/lib${DYLD_LIBRARY_PATH:+:$DYLD_LIBRARY_PATH}"
+  else
+    export LD_LIBRARY_PATH="$ROOT_DIR/cgo:$ROOT_DIR/thirdparties/install/lib${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}"
+  fi
+  CURRENT_STAGE="start_matrixone"
+  "$service_bin" -launch "$TMP_DIR/mo-config/launch.toml" >"$TMP_DIR/mo-service.log" 2>&1 &
+  MO_PID=$!
+  mo_e2e_wait_frontend_port "$MO_PID" "$TMP_DIR/mo-service.log" "$MO_PORT" >/dev/null || die "MatrixOne did not bind the owned frontend port"
+
+  CURRENT_STAGE="validate_compose"
+  compose config -q
+  CURRENT_STAGE="start_kafka_connect"
+  compose up -d
+  CURRENT_STAGE="run_recovery_driver"
+  local binary_sha
+  binary_sha="$(shasum -a 256 "$service_bin" | awk '{print $1}')"
+  (cd "$ROOT_DIR" && go run ./test/kafkaconnect/kafka_connect_e2e_local.go \
+    --dsn "root:111@tcp(127.0.0.1:$MO_PORT)/?timeout=5s&readTimeout=30s&writeTimeout=30s" \
+    --bootstrap "127.0.0.1:$KAFKA_PORT" \
+    --connect-url "http://127.0.0.1:$CONNECT_PORT" \
+    --compose-base "$COMPOSE_BASE" --compose-connect "$COMPOSE_CONNECT" \
+    --compose-project "$COMPOSE_PROJECT_NAME" --connector-name "mo-issue-29848-recovery" \
+    --source-sha "$SOURCE_SHA" --harness-sha "$HARNESS_SHA" \
+    --mo-binary-sha "$binary_sha" --report-dir "$REPORT_DIR")
+}
+
+case "$PROFILE" in
+  e2e-local) run_e2e ;;
+  *) die "unknown profile: $PROFILE" ;;
+esac
diff --git a/optools/mo_e2e_common.bash b/optools/mo_e2e_common.bash
new file mode 100644
index 0000000000000..0e0d8663b16bf
--- /dev/null
+++ b/optools/mo_e2e_common.bash
@@ -0,0 +1,191 @@
+#!/usr/bin/env bash
+# Copyright 2026 Matrix Origin
+# Licensed under the Apache License, Version 2.0.
+
+# Shared host-wide port-block ownership for local integration runners.
+# Callers own PORT_LEASE_PID and must keep it alive until their services stop.
+
+mo_port_block_candidates() {
+  local width="${1:-80}"
+  python3 - "$width" <<'PY'
+import random
+import sys
+
+width = int(sys.argv[1])
+candidates = list(range(36000, 57000 - width, width))
+random.shuffle(candidates)
+for base in candidates:
+    print(base)
+PY
+}
+
+mo_port_block_is_available() {
+  local base="$1" width="${2:-80}"
+  python3 - "$base" "$width" <<'PY'
+import socket
+import sys
+
+base = int(sys.argv[1])
+width = int(sys.argv[2])
+sockets = []
+try:
+    for port in range(base, base + width):
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.bind(("127.0.0.1", port))
+        sockets.append(sock)
+except OSError:
+    raise SystemExit(1)
+finally:
+    for sock in sockets:
+        sock.close()
+PY
+}
+
+mo_port_block_acquire_lease() {
+  local base="$1" lease_dir="$2" tmp_root="${3:-${TMPDIR:-/tmp}}"
+  local status_file status=""
+  mkdir -p "$lease_dir" || return 2
+  chmod 700 "$lease_dir" || return 2
+  local lease_file="$lease_dir/$base.lock"
+  status_file="$(mktemp "${tmp_root%/}/mo-port-block-lease.XXXXXX")" || return 2
+
+  python3 - "$lease_file" "$status_file" "$$" <<'PY' &
+import fcntl
+import os
+import signal
+import sys
+import time
+
+lease_file, status_file, parent_pid = sys.argv[1:]
+parent_pid = int(parent_pid)
+with open(lease_file, "a+") as lease:
+    try:
+        fcntl.flock(lease, fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except BlockingIOError:
+        with open(status_file, "w") as status:
+            status.write("busy\n")
+        raise SystemExit(1)
+    with open(status_file, "w") as status:
+        status.write("locked\n")
+
+    def release(_signal, _frame):
+        raise SystemExit(0)
+
+    signal.signal(signal.SIGTERM, release)
+    signal.signal(signal.SIGINT, release)
+    while os.getppid() == parent_pid:
+        time.sleep(0.05)
+PY
+  PORT_LEASE_PID=$!
+
+  for _ in {1..100}; do
+    if [[ -s "$status_file" ]]; then
+      status="$(cat "$status_file")"
+      break
+    fi
+    if ! kill -0 "$PORT_LEASE_PID" >/dev/null 2>&1; then
+      break
+    fi
+    sleep 0.05
+  done
+  rm -f -- "$status_file"
+
+  if [[ "$status" == "locked" ]]; then
+    return 0
+  fi
+  wait "$PORT_LEASE_PID" >/dev/null 2>&1 || true
+  PORT_LEASE_PID=""
+  return 1
+}
+
+mo_port_block_release_lease() {
+  if [[ -n "${PORT_LEASE_PID:-}" ]]; then
+    kill "$PORT_LEASE_PID" >/dev/null 2>&1 || true
+    wait "$PORT_LEASE_PID" >/dev/null 2>&1 || true
+  fi
+  PORT_LEASE_PID=""
+}
+
+mo_e2e_generate_launch_config() {
+  local root_dir="$1" tmp_dir="$2" log_port_base="$3" frontend_port="$4" status_port="$5"
+  local source_dir="$root_dir/etc/launch" generated_dir="$tmp_dir/mo-config"
+  local log_service_port=$((log_port_base + 1)) log_gossip_port=$((log_port_base + 2))
+  local tn_port_base=$((log_port_base + 24)) cn_port_base=$((log_port_base + 48))
+  mkdir -p "$generated_dir"
+  local name
+  for name in log tn; do
+    sed -e "s#\./mo-data#$tmp_dir/mo-data#g" -e "s#\"mo-data/#\"$tmp_dir/mo-data/#g" \
+      -e "s#^status-port = [0-9][0-9]*#status-port = $status_port#" \
+      -e "s#^port-base = [0-9][0-9]*#port-base = $tn_port_base#" \
+      "$source_dir/$name.toml" >"$generated_dir/$name.toml"
+  done
+  sed -e "s#\./mo-data#$tmp_dir/mo-data#g" -e "s#\"mo-data/#\"$tmp_dir/mo-data/#g" \
+    -e "s#^status-port = [0-9][0-9]*#status-port = $status_port#" \
+    -e "s#^port-base = [0-9][0-9]*#port-base = $cn_port_base#" \
+    "$source_dir/cn.toml" | awk -v port="$frontend_port" -v socket="$tmp_dir/mysql.sock" '
+      /^\[cn\.frontend\.iceberg\]$/ && !inserted {
+        print "[cn.frontend]"
+        print "port = " port
+        print "unix-socket = \"" socket "\""
+        print ""
+        inserted = 1
+      }
+      { print }
+      END { if (!inserted) exit 42 }
+    ' >"$generated_dir/cn.toml"
+  {
+    printf '\n[logservice]\n'
+    printf 'raft-port = %s\n' "$log_port_base"
+    printf 'logservice-port = %s\n' "$log_service_port"
+    printf 'gossip-port = %s\n' "$log_gossip_port"
+    printf 'gossip-seed-addresses = ["127.0.0.1:%s"]\n' "$log_gossip_port"
+  } >>"$generated_dir/log.toml"
+  for name in log tn cn; do
+    {
+      printf '\n[hakeeper-client]\n'
+      printf 'service-addresses = ["127.0.0.1:%s"]\n' "$log_service_port"
+    } >>"$generated_dir/$name.toml"
+  done
+  sed -e "s#\./etc/launch/log.toml#$generated_dir/log.toml#" \
+    -e "s#\./etc/launch/tn.toml#$generated_dir/tn.toml#" \
+    -e "s#\./etc/launch/cn.toml#$generated_dir/cn.toml#" \
+    "$source_dir/launch.toml" >"$generated_dir/launch.toml"
+}
+
+mo_e2e_wait_frontend_port() {
+  local pid="$1" log_file="$2" expected_port="${3:-}" deadline=$((SECONDS + 120)) port=""
+  while (( SECONDS < deadline )); do
+    if [[ -n "$pid" ]] && ! kill -0 "$pid" >/dev/null 2>&1; then
+      printf 'MatrixOne exited before publishing its frontend listener\n' >&2
+      return 1
+    fi
+    port="$(sed -nE 's/.*Server Listening on : [^ ]*:([0-9]+).*/\1/p' "$log_file" | tail -1)"
+    if [[ "$port" =~ ^[1-9][0-9]*$ ]]; then
+      if [[ -n "$expected_port" && "$port" != "$expected_port" ]]; then
+        printf 'MatrixOne published frontend port %s, expected %s\n' "$port" "$expected_port" >&2
+        return 1
+      fi
+      printf '%s\n' "$port"
+      return 0
+    fi
+    sleep 0.25
+  done
+  printf 'MatrixOne did not publish its frontend listener before timeout\n' >&2
+  return 1
+}
+
+mo_e2e_stop_service() {
+  local pid="$1"
+  [[ -n "$pid" ]] || return 0
+  if kill -0 "$pid" >/dev/null 2>&1; then
+    kill "$pid" >/dev/null 2>&1 || true
+    for _ in {1..40}; do
+      kill -0 "$pid" >/dev/null 2>&1 || break
+      sleep 0.25
+    done
+    if kill -0 "$pid" >/dev/null 2>&1; then
+      kill -KILL "$pid" >/dev/null 2>&1 || true
+    fi
+    wait "$pid" >/dev/null 2>&1 || true
+  fi
+}
diff --git a/optools/mongodb_ci.bash b/optools/mongodb_ci.bash
index 410d2f1a73d82..b6aaa4433e9c8 100755
--- a/optools/mongodb_ci.bash
+++ b/optools/mongodb_ci.bash
@@ -22,6 +22,8 @@ readonly CN_PORT_OFFSET=48
 readonly FRONTEND_PORT_OFFSET=72
 readonly STATUS_PORT_OFFSET=73
 
+source "$ROOT_DIR/optools/mo_e2e_common.bash"
+
 log() { printf '[mongodb-ci] %s\n' "$*"; }
 die() { printf '[mongodb-ci] ERROR: %s\n' "$*" >&2; exit 1; }
 require() { command -v "$1" >/dev/null 2>&1 || die "required command not found: $1"; }
@@ -66,17 +68,8 @@ cleanup() {
   # Cleanup must continue even when artifact collection itself fails. Preserve
   # the original test status and make every teardown action best-effort.
   set +e
-  if [[ -n "$MO_PID" ]] && kill -0 "$MO_PID" >/dev/null 2>&1; then
-    kill "$MO_PID" >/dev/null 2>&1 || true
-    for _ in {1..40}; do
-      kill -0 "$MO_PID" >/dev/null 2>&1 || break
-      sleep 0.25
-    done
-    if kill -0 "$MO_PID" >/dev/null 2>&1; then
-      kill -KILL "$MO_PID" >/dev/null 2>&1 || true
-    fi
-    wait "$MO_PID" >/dev/null 2>&1 || true
-  fi
+  mo_e2e_stop_service "$MO_PID"
+  MO_PID=""
   collect
   if [[ -n "$TMP_DIR" ]]; then
     MONGODB_PORT="${MONGODB_PORT:-27017}" MONGODB_ROOT_USER="${MONGODB_ROOT_USER:-x}" \
@@ -114,19 +107,8 @@ wait_mongo() {
 }
 
 wait_mo_port() {
-  local deadline=$((SECONDS + 120)) port=""
-  while (( SECONDS < deadline )); do
-    if [[ -n "$MO_PID" ]] && ! kill -0 "$MO_PID" >/dev/null 2>&1; then
-      die "MatrixOne exited before publishing its frontend listener"
-    fi
-    port="$(sed -nE 's/.*Server Listening on : [^ ]*:([0-9]+).*/\1/p' "$TMP_DIR/mo-service.log" | tail -1)"
-    if [[ "$port" =~ ^[1-9][0-9]*$ ]]; then
-      printf '%s\n' "$port"
-      return
-    fi
-    sleep 0.25
-  done
-  die "MatrixOne did not publish its frontend listener"
+  mo_e2e_wait_frontend_port "$MO_PID" "$TMP_DIR/mo-service.log" || \
+    die "MatrixOne did not publish its frontend listener"
 }
 
 wait_primary() {
@@ -140,109 +122,25 @@ wait_primary() {
 }
 
 port_block_candidates() {
-  python3 - <<'PY'
-import random
-
-# Keep every MatrixOne listener in one 80-port block. Availability is checked
-# after a host-wide lease has been acquired for each candidate.
-width = 80
-candidates = list(range(36000, 57000 - width, width))
-random.shuffle(candidates)
-for base in candidates:
-    print(base)
-PY
+  mo_port_block_candidates "$PORT_BLOCK_WIDTH"
 }
 
 port_block_is_available() {
-  local base="$1"
-  python3 - "$base" "$PORT_BLOCK_WIDTH" <<'PY'
-import socket
-import sys
-
-base = int(sys.argv[1])
-width = int(sys.argv[2])
-sockets = []
-try:
-    for port in range(base, base + width):
-        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        sock.bind(("127.0.0.1", port))
-        sockets.append(sock)
-except OSError:
-    raise SystemExit(1)
-finally:
-    for sock in sockets:
-        sock.close()
-PY
+  mo_port_block_is_available "$1" "$PORT_BLOCK_WIDTH"
 }
 
 acquire_port_block_lease() {
-  local base="$1" lease_dir status_file status=""
-  # TMPDIR may be scoped to an individual CI job. The default lock directory
-  # must instead be stable for every process running under this host user.
-  lease_dir="${MO_MONGODB_PORT_LEASE_DIR:-/tmp/mo-mongodb-port-leases-$(id -u)}"
-  mkdir -p "$lease_dir" || die "could not create MongoDB E2E port lease directory"
-  chmod 700 "$lease_dir" || die "could not secure MongoDB E2E port lease directory"
-  local lease_file="$lease_dir/$base.lock"
-  status_file="$(mktemp "${TMPDIR:-/tmp}/mo-mongodb-port-lease.XXXXXX")" || \
-    die "could not create MongoDB E2E port lease status file"
-
-  python3 - "$lease_file" "$status_file" "$$" <<'PY' &
-import fcntl
-import os
-import signal
-import sys
-import time
-
-lease_file, status_file, parent_pid = sys.argv[1:]
-parent_pid = int(parent_pid)
-with open(lease_file, "a+") as lease:
-    try:
-        fcntl.flock(lease, fcntl.LOCK_EX | fcntl.LOCK_NB)
-    except BlockingIOError:
-        with open(status_file, "w") as status:
-            status.write("busy\n")
-        raise SystemExit(1)
-    with open(status_file, "w") as status:
-        status.write("locked\n")
-
-    def release(_signal, _frame):
-        raise SystemExit(0)
-
-    signal.signal(signal.SIGTERM, release)
-    signal.signal(signal.SIGINT, release)
-    # The shell trap handles ordinary cleanup. If the shell is killed, the
-    # child is reparented and must close its flock without waiting for a signal.
-    while os.getppid() == parent_pid:
-        time.sleep(0.05)
-PY
-  PORT_LEASE_PID=$!
-
-  for _ in {1..100}; do
-    if [[ -s "$status_file" ]]; then
-      status="$(cat "$status_file")"
-      break
-    fi
-    if ! kill -0 "$PORT_LEASE_PID" >/dev/null 2>&1; then
-      break
-    fi
-    sleep 0.05
-  done
-  rm -f -- "$status_file"
-
-  if [[ "$status" == "locked" ]]; then
-    return
+  local lease_dir="${MO_MONGODB_PORT_LEASE_DIR:-/tmp/mo-mongodb-port-leases-$(id -u)}"
+  local status=0
+  mo_port_block_acquire_lease "$1" "$lease_dir" "${TMPDIR:-/tmp}" || status=$?
+  if (( status == 2 )); then
+    die "could not create or secure MongoDB E2E port-block lease"
   fi
-  wait "$PORT_LEASE_PID" >/dev/null 2>&1 || true
-  PORT_LEASE_PID=""
-  return 1
+  return "$status"
 }
 
 release_port_block_lease() {
-  if [[ -n "$PORT_LEASE_PID" ]]; then
-    kill "$PORT_LEASE_PID" >/dev/null 2>&1 || true
-    wait "$PORT_LEASE_PID" >/dev/null 2>&1 || true
-  fi
-  PORT_LEASE_PID=""
+  mo_port_block_release_lease
 }
 
 is_tcp_port() {
@@ -392,57 +290,14 @@ print_port_plan() {
 }
 
 generate_mo_config() {
-  local source_dir="$ROOT_DIR/etc/launch"
   local generated_dir="$TMP_DIR/mo-config"
-  mkdir -p "$generated_dir"
-  for name in log tn; do
-    sed -e "s#\./mo-data#$TMP_DIR/mo-data#g" -e "s#\"mo-data/#\"$TMP_DIR/mo-data/#g" \
-      -e "s#^status-port = [0-9][0-9]*#status-port = $STATUS_PORT#" \
-      -e "s#^port-base = [0-9][0-9]*#port-base = $TN_PORT_BASE#" \
-      "$source_dir/$name.toml" >"$generated_dir/$name.toml"
-  done
-  sed -e "s#\./mo-data#$TMP_DIR/mo-data#g" -e "s#\"mo-data/#\"$TMP_DIR/mo-data/#g" \
-      -e "s#^status-port = [0-9][0-9]*#status-port = $STATUS_PORT#" \
-      -e "s#^port-base = [0-9][0-9]*#port-base = $CN_PORT_BASE#" \
-      "$source_dir/cn.toml" | awk -v port="$MO_PORT" '
-        /^\[cn\.frontend\.iceberg\]$/ && !inserted {
-          print "[cn.frontend]"
-          print "port = " port
-          print ""
-          inserted = 1
-        }
-        { print }
-        END { if (!inserted) exit 42 }
-      ' >"$generated_dir/cn.toml"
-  {
-    printf '\n[logservice]\n'
-    printf 'raft-port = %s\n' "$LOG_RAFT_PORT"
-    printf 'logservice-port = %s\n' "$LOG_SERVICE_PORT"
-    printf 'gossip-port = %s\n' "$LOG_GOSSIP_PORT"
-    printf 'gossip-seed-addresses = ["127.0.0.1:%s"]\n' "$LOG_GOSSIP_PORT"
-  } >>"$generated_dir/log.toml"
-  {
-    printf '\n[hakeeper-client]\n'
-    printf 'service-addresses = ["127.0.0.1:%s"]\n' "$LOG_SERVICE_PORT"
-  } >>"$generated_dir/log.toml"
-  {
-    printf '\n[hakeeper-client]\n'
-    printf 'service-addresses = ["127.0.0.1:%s"]\n' "$LOG_SERVICE_PORT"
-  } >>"$generated_dir/tn.toml"
-  {
-    printf '\n[hakeeper-client]\n'
-    printf 'service-addresses = ["127.0.0.1:%s"]\n' "$LOG_SERVICE_PORT"
-  } >>"$generated_dir/cn.toml"
+  mo_e2e_generate_launch_config "$ROOT_DIR" "$TMP_DIR" "$LOG_PORT_BASE" "$MO_PORT" "$STATUS_PORT"
   {
     printf '\n[cn.frontend.mongodb]\n'
     printf 'enable = true\nallow-loopback = true\n'
     printf 'connect-timeout = "10s"\nserver-selection-timeout = "10s"\nsocket-timeout = "30s"\n'
     printf 'batch-rows = 2\nmax-batch-bytes = 1048576\nmax-value-bytes = 524288\nmax-source-concurrency = 2\n'
   } >>"$generated_dir/cn.toml"
-  sed -e "s#\./etc/launch/log.toml#$generated_dir/log.toml#" \
-      -e "s#\./etc/launch/tn.toml#$generated_dir/tn.toml#" \
-      -e "s#\./etc/launch/cn.toml#$generated_dir/cn.toml#" \
-      "$source_dir/launch.toml" >"$generated_dir/launch.toml"
 }
 
 run_e2e() {
diff --git a/test/kafkaconnect/README.md b/test/kafkaconnect/README.md
new file mode 100644
index 0000000000000..2db049a8487dc
--- /dev/null
+++ b/test/kafkaconnect/README.md
@@ -0,0 +1,28 @@
+# Kafka Connect recovery qualification
+
+`optools/kafka_connect_ci.bash e2e-local` runs one real Kafka 3.9.0 / Kafka
+Connect 3.9.0 / Debezium JDBC 3.1.2 sink path against a locally built MatrixOne. It
+requires Docker Compose, Go, Python 3, Make, and the MatrixOne native build
+dependencies. Images are pinned by digest in `compose.connect.yaml`.
+
+The run checks a baseline insert, kills the owned Connect worker, writes an
+insert, update, and keyed tombstone while it is down, then starts the same
+worker and checks the exact destination rows plus the Connect REST offset,
+Kafka consumer-group committed offset, and Kafka log-end offset. It also stops
+the connector, rewinds the fixture consumer offset through Kafka Connect's REST
+API, verifies replay leaves the target unchanged, and checks an empty-work
+worker restart. The report records the
+MatrixOne binary source revision, harness source revision, binary digest,
+Connect/plugin versions, exact rows, and offsets under
+`test/kafkaconnect/reports/` by default. Set
+`MO_KAFKA_CONNECT_REPORT_DIR` to select another report directory.
+The driver limits the full run to 10 minutes and each readiness or drain wait
+to 90 seconds by default; `--timeout` and `--phase-timeout` can adjust those
+limits. It atomically updates `report.json` after each stage so an interrupted
+run retains its current stage and completed phase receipts. Artifact collection
+or service-cleanup failures change the overall report to `failed` and make the
+runner exit nonzero.
+
+This harness qualifies only this Kafka Connect JDBC sink recovery path. It does
+not certify Airbyte, NiFi generic CDC, large-volume recovery, or cross-system
+exactly-once delivery.
diff --git a/test/kafkaconnect/connect_client_test.go b/test/kafkaconnect/connect_client_test.go
new file mode 100644
index 0000000000000..cf66681cf68f5
--- /dev/null
+++ b/test/kafkaconnect/connect_client_test.go
@@ -0,0 +1,87 @@
+// Copyright 2026 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+
+package main
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestConnectClientOffsets(t *testing.T) {
+	t.Run("read requested topic partition", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if r.Method != http.MethodGet || r.URL.Path != "/connectors/fixture/offsets" {
+				t.Errorf("request = %s %s, want GET /connectors/fixture/offsets", r.Method, r.URL.Path)
+			}
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = w.Write([]byte(`{"offsets":[{"partition":{"kafka_topic":"other","kafka_partition":0},"offset":{"kafka_offset":2}},{"partition":{"kafka_topic":"mo_issue_29848_events","kafka_partition":0},"offset":{"kafka_offset":17}}]}`))
+		}))
+		defer server.Close()
+
+		got, err := (connectClient{base: server.URL, client: server.Client()}).restOffset(t.Context(), "fixture", topicName, 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if got != 17 {
+			t.Fatalf("offset = %d, want 17", got)
+		}
+	})
+
+	t.Run("reject missing partition offset", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			_, _ = w.Write([]byte(`{"offsets":[{"partition":{"kafka_topic":"mo_issue_29848_events","kafka_partition":0},"offset":null}]}`))
+		}))
+		defer server.Close()
+
+		_, err := (connectClient{base: server.URL, client: server.Client()}).restOffset(t.Context(), "fixture", topicName, 0)
+		if err == nil || !strings.Contains(err.Error(), "omit "+topicName+"[0]") {
+			t.Fatalf("restOffset error = %v, want missing partition offset", err)
+		}
+	})
+
+	t.Run("patch only the requested partition and offset", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if r.Method != http.MethodPatch || r.URL.Path != "/connectors/fixture/offsets" {
+				t.Errorf("request = %s %s, want PATCH /connectors/fixture/offsets", r.Method, r.URL.Path)
+			}
+			if got := r.Header.Get("Content-Type"); got != "application/json" {
+				t.Errorf("Content-Type = %q, want application/json", got)
+			}
+			var body connectorOffsets
+			if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+				t.Errorf("decode PATCH body: %v", err)
+				w.WriteHeader(http.StatusBadRequest)
+				return
+			}
+			if len(body.Offsets) != 1 || body.Offsets[0].Partition.KafkaTopic != topicName || body.Offsets[0].Partition.KafkaPartition != 0 || body.Offsets[0].Offset == nil || body.Offsets[0].Offset.KafkaOffset != 9 {
+				t.Errorf("PATCH body = %+v, want only %s[0] offset 9", body, topicName)
+				w.WriteHeader(http.StatusBadRequest)
+				return
+			}
+			w.WriteHeader(http.StatusAccepted)
+		}))
+		defer server.Close()
+
+		if err := (connectClient{base: server.URL, client: server.Client()}).patchOffset(t.Context(), "fixture", topicName, 0, 9); err != nil {
+			t.Fatal(err)
+		}
+	})
+
+	t.Run("reject unexpected patch response", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			http.Error(w, "connector must be stopped", http.StatusConflict)
+		}))
+		defer server.Close()
+
+		err := (connectClient{base: server.URL, client: server.Client()}).patchOffset(t.Context(), "fixture", topicName, 0, 9)
+		if err == nil || !strings.Contains(err.Error(), "HTTP 409") {
+			t.Fatalf("patchOffset error = %v, want HTTP 409", err)
+		}
+	})
+}
diff --git a/test/kafkaconnect/kafka_connect_e2e_local.go b/test/kafkaconnect/kafka_connect_e2e_local.go
new file mode 100644
index 0000000000000..e7da2727739d3
--- /dev/null
+++ b/test/kafkaconnect/kafka_connect_e2e_local.go
@@ -0,0 +1,1224 @@
+// Copyright 2026 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+
+// Command kafka_connect_e2e_local qualifies one real Kafka Connect JDBC sink
+// recovery path against an owned MatrixOne, Kafka broker, and Connect worker.
+package main
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"reflect"
+	"sort"
+	"strings"
+	"time"
+
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/twmb/franz-go/pkg/kadm"
+	"github.com/twmb/franz-go/pkg/kgo"
+)
+
+const (
+	topicName      = "mo_issue_29848_events"
+	databaseName   = "kafka_connect_recovery"
+	tableName      = "events"
+	connectorClass = "io.debezium.connector.jdbc.JdbcSinkConnector"
+)
+
+var errDuplicateRowID = errors.New("duplicate expected row ID")
+
+type row struct {
+	ID      int64
+	Name    string
+	Note    *string
+	Counter int64
+}
+
+type producedRecord struct {
+	Partition int32 `json:"partition"`
+	Offset    int64 `json:"offset"`
+	ID        int64 `json:"id"`
+	Tombstone bool  `json:"tombstone"`
+}
+
+type phase struct {
+	Name              string           `json:"name"`
+	Status            string           `json:"status"`
+	ExpectedOffset    int64            `json:"expected_offset"`
+	RESTOffset        int64            `json:"rest_offset"`
+	KafkaGroupOffset  int64            `json:"kafka_group_offset"`
+	KafkaLogEndOffset int64            `json:"kafka_log_end_offset"`
+	Produced          []producedRecord `json:"produced,omitempty"`
+	Rows              []rowReport      `json:"rows,omitempty"`
+	ConnectorState    string           `json:"connector_state,omitempty"`
+	WorkerRestarted   bool             `json:"worker_restarted,omitempty"`
+	ControlledReplay  bool             `json:"controlled_replay,omitempty"`
+	Detail            string           `json:"detail,omitempty"`
+}
+
+type rowReport struct {
+	ID      int64   `json:"id"`
+	Name    string  `json:"name"`
+	Note    *string `json:"note"`
+	Counter int64   `json:"counter"`
+}
+
+type runReport struct {
+	Status            string  `json:"status"`
+	CurrentStage      string  `json:"current_stage,omitempty"`
+	StartedAt         string  `json:"started_at"`
+	FinishedAt        string  `json:"finished_at,omitempty"`
+	SourceSHA         string  `json:"source_sha"`
+	HarnessSHA        string  `json:"harness_sha"`
+	MOBinarySHA256    string  `json:"mo_binary_sha256"`
+	KafkaBootstrap    string  `json:"kafka_bootstrap"`
+	ConnectURL        string  `json:"connect_url"`
+	ConnectorName     string  `json:"connector_name"`
+	SinkConsumerGroup string  `json:"sink_consumer_group"`
+	ConnectVersion    string  `json:"connect_version,omitempty"`
+	KafkaClusterID    string  `json:"kafka_cluster_id,omitempty"`
+	JDBCPluginVersion string  `json:"jdbc_plugin_version,omitempty"`
+	Phases            []phase `json:"phases"`
+	Error             string  `json:"error,omitempty"`
+}
+
+type connectRoot struct {
+	Version        string `json:"version"`
+	Commit         string `json:"commit"`
+	KafkaClusterID string `json:"kafka_cluster_id"`
+}
+
+type connectorPlugin struct {
+	Class   string `json:"class"`
+	Type    string `json:"type"`
+	Version string `json:"version"`
+}
+
+type connectorStatus struct {
+	Name   string `json:"name"`
+	State  string `json:"state"`
+	Worker string `json:"worker_id"`
+}
+
+type taskStatus struct {
+	ID     int    `json:"id"`
+	State  string `json:"state"`
+	Worker string `json:"worker_id"`
+	Trace  string `json:"trace,omitempty"`
+}
+
+type connectorStatusResponse struct {
+	Connector connectorStatus `json:"connector"`
+	Tasks     []taskStatus    `json:"tasks"`
+}
+
+type offsetPartition struct {
+	KafkaTopic     string `json:"kafka_topic"`
+	KafkaPartition int32  `json:"kafka_partition"`
+}
+
+type offsetValue struct {
+	KafkaOffset int64 `json:"kafka_offset"`
+}
+
+type connectorOffset struct {
+	Partition offsetPartition `json:"partition"`
+	Offset    *offsetValue    `json:"offset"`
+}
+
+type connectorOffsets struct {
+	Offsets []connectorOffset `json:"offsets"`
+}
+
+type connectClient struct {
+	base   string
+	client *http.Client
+}
+
+type drainReceipt struct {
+	RESTOffset        int64
+	KafkaGroupOffset  int64
+	KafkaLogEndOffset int64
+}
+
+type drainStateReader interface {
+	rows(context.Context) ([]row, error)
+	restOffset(context.Context) (int64, error)
+	groupOffset(context.Context) (int64, bool, error)
+	logEndOffset(context.Context) (int64, bool, error)
+}
+
+type connectorDrainReader struct {
+	db      *sql.DB
+	admin   *kadm.Client
+	connect *connectClient
+	name    string
+}
+
+func (r connectorDrainReader) rows(ctx context.Context) ([]row, error) {
+	return readRows(ctx, r.db)
+}
+
+func (r connectorDrainReader) restOffset(ctx context.Context) (int64, error) {
+	return r.connect.restOffset(ctx, r.name, topicName, 0)
+}
+
+func (r connectorDrainReader) groupOffset(ctx context.Context) (int64, bool, error) {
+	return groupOffset(ctx, r.admin, "connect-"+r.name)
+}
+
+func (r connectorDrainReader) logEndOffset(ctx context.Context) (int64, bool, error) {
+	return topicLogEndOffset(ctx, r.admin)
+}
+
+type recoveryRuntime interface {
+	drainStateReader
+	prepareTarget(context.Context) error
+	createTopic(context.Context) error
+	produceRows(context.Context, []row) ([]producedRecord, error)
+	produceTombstone(context.Context, int64) ([]producedRecord, error)
+	killWorker(context.Context) error
+	startWorker(context.Context) error
+}
+
+type liveRecoveryRuntime struct {
+	connectorDrainReader
+	kafka *kgo.Client
+	opts  runOptions
+}
+
+func (r liveRecoveryRuntime) prepareTarget(ctx context.Context) error {
+	return prepareTarget(ctx, r.db)
+}
+
+func (r liveRecoveryRuntime) createTopic(ctx context.Context) error {
+	return createTopic(ctx, r.admin)
+}
+
+func (r liveRecoveryRuntime) produceRows(ctx context.Context, rows []row) ([]producedRecord, error) {
+	return produceRows(ctx, r.kafka, rows)
+}
+
+func (r liveRecoveryRuntime) produceTombstone(ctx context.Context, id int64) ([]producedRecord, error) {
+	return produceTombstone(ctx, r.kafka, id)
+}
+
+func (r liveRecoveryRuntime) killWorker(ctx context.Context) error {
+	return killOwnedConnect(ctx, r.opts)
+}
+
+func (r liveRecoveryRuntime) startWorker(ctx context.Context) error {
+	return startOwnedConnect(ctx, r.opts)
+}
+
+type runOptions struct {
+	dsn, bootstrap, connectURL, composeBase, composeConnect, composeProject string
+	connectorName, sourceSHA, harnessSHA, moBinarySHA, reportDir            string
+	totalTimeout, phaseTimeout                                              time.Duration
+}
+
+type runDependencies struct {
+	openDatabase    func(string) (*sql.DB, error)
+	newKafkaClient  func(...kgo.Opt) (*kgo.Client, error)
+	runRecoveryFunc func(context.Context, runOptions, *runReport, connectClient, recoveryRuntime) error
+}
+
+func defaultRunDependencies() runDependencies {
+	return runDependencies{
+		openDatabase:    func(dsn string) (*sql.DB, error) { return sql.Open("mysql", dsn) },
+		newKafkaClient:  kgo.NewClient,
+		runRecoveryFunc: runRecovery,
+	}
+}
+
+func main() {
+	os.Exit(runMain(os.Args[1:]))
+}
+
+func runMain(args []string) int {
+	var opts runOptions
+	flags := flag.NewFlagSet("kafka_connect_e2e_local", flag.ContinueOnError)
+	flags.SetOutput(os.Stderr)
+	flags.StringVar(&opts.dsn, "dsn", "root:111@tcp(127.0.0.1:6001)/?timeout=5s&readTimeout=30s&writeTimeout=30s", "owned MatrixOne MySQL DSN")
+	flags.StringVar(&opts.bootstrap, "bootstrap", "127.0.0.1:9092", "owned Kafka broker bootstrap address")
+	flags.StringVar(&opts.connectURL, "connect-url", "http://127.0.0.1:8083", "owned Kafka Connect REST URL")
+	flags.StringVar(&opts.composeBase, "compose-base", "etc/launch-kafka-local/compose.yaml", "base Kafka Compose file")
+	flags.StringVar(&opts.composeConnect, "compose-connect", "etc/launch-kafka-local/compose.connect.yaml", "Connect Compose file")
+	flags.StringVar(&opts.composeProject, "compose-project", "", "owned Compose project")
+	flags.StringVar(&opts.connectorName, "connector-name", "mo-issue-29848-recovery", "owned sink connector name")
+	flags.StringVar(&opts.sourceSHA, "source-sha", "unknown", "MatrixOne source revision")
+	flags.StringVar(&opts.harnessSHA, "harness-sha", "unknown", "harness source revision")
+	flags.StringVar(&opts.moBinarySHA, "mo-binary-sha", "unknown", "built MatrixOne binary SHA-256")
+	flags.StringVar(&opts.reportDir, "report-dir", "test/kafkaconnect/reports/local", "evidence output directory")
+	flags.DurationVar(&opts.totalTimeout, "timeout", 10*time.Minute, "maximum duration for the complete recovery run")
+	flags.DurationVar(&opts.phaseTimeout, "phase-timeout", 90*time.Second, "maximum duration for each readiness or drain phase")
+	if err := flags.Parse(args); err != nil {
+		if errors.Is(err, flag.ErrHelp) {
+			return 0
+		}
+		return 2
+	}
+	if opts.totalTimeout <= 0 || opts.phaseTimeout <= 0 {
+		fmt.Fprintln(os.Stderr, "timeout values must be positive")
+		return 2
+	}
+
+	report := runReport{
+		Status:            "running",
+		StartedAt:         time.Now().UTC().Format(time.RFC3339Nano),
+		SourceSHA:         opts.sourceSHA,
+		HarnessSHA:        opts.harnessSHA,
+		MOBinarySHA256:    opts.moBinarySHA,
+		KafkaBootstrap:    opts.bootstrap,
+		ConnectURL:        opts.connectURL,
+		ConnectorName:     opts.connectorName,
+		SinkConsumerGroup: "connect-" + opts.connectorName,
+		Phases:            []phase{},
+	}
+	if err := writeReport(opts.reportDir, report); err != nil {
+		fmt.Fprintln(os.Stderr, fmt.Errorf("write initial evidence report: %w", err))
+		return 1
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), opts.totalTimeout)
+	defer cancel()
+
+	err := run(ctx, opts, &report)
+	setRunResult(&report, err)
+	report.FinishedAt = time.Now().UTC().Format(time.RFC3339Nano)
+	if writeErr := writeReport(opts.reportDir, report); writeErr != nil {
+		if err == nil {
+			err = fmt.Errorf("write evidence report: %w", writeErr)
+		} else {
+			err = errors.Join(err, fmt.Errorf("write evidence report: %w", writeErr))
+		}
+	}
+	if err == nil {
+		fmt.Printf("Kafka Connect recovery report: %s\n", filepath.Join(opts.reportDir, "report.json"))
+	}
+	if err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		return 1
+	}
+	return 0
+}
+
+func setRunResult(report *runReport, runErr error) {
+	if runErr != nil {
+		report.Status = "failed"
+		report.Error = runErr.Error()
+		return
+	}
+	report.Status = "passed"
+}
+
+func run(ctx context.Context, opts runOptions, report *runReport, provided ...runDependencies) error {
+	dependencies := defaultRunDependencies()
+	if len(provided) > 0 {
+		dependencies = provided[0]
+	}
+	if opts.composeProject == "" {
+		return errors.New("compose project is required to operate only on run-owned containers")
+	}
+	connect := connectClient{base: strings.TrimRight(opts.connectURL, "/"), client: &http.Client{Timeout: 10 * time.Second}}
+	db, err := dependencies.openDatabase(opts.dsn)
+	if err != nil {
+		return fmt.Errorf("open MatrixOne DSN: %w", err)
+	}
+	defer db.Close()
+	db.SetMaxOpenConns(1)
+	if err := setStage(opts, report, "wait-for-matrixone"); err != nil {
+		return err
+	}
+	phaseCtx, phaseCancel := context.WithTimeout(ctx, opts.phaseTimeout)
+	err = waitForDatabase(phaseCtx, db)
+	phaseCancel()
+	if err != nil {
+		return err
+	}
+	if err := setStage(opts, report, "wait-for-connect"); err != nil {
+		return err
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	root, err := waitForConnect(phaseCtx, connect)
+	phaseCancel()
+	if err != nil {
+		return err
+	}
+	report.ConnectVersion = root.Version
+	report.KafkaClusterID = root.KafkaClusterID
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	plugins, err := connect.plugins(phaseCtx)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("list Kafka Connect plugins: %w", err)
+	}
+	pluginVersion := ""
+	for _, plugin := range plugins {
+		if plugin.Class == connectorClass && plugin.Type == "sink" {
+			pluginVersion = plugin.Version
+			break
+		}
+	}
+	if pluginVersion == "" {
+		return fmt.Errorf("required sink plugin %q is absent from /connector-plugins", connectorClass)
+	}
+	report.JDBCPluginVersion = pluginVersion
+	kafka, err := dependencies.newKafkaClient(kgo.SeedBrokers(opts.bootstrap), kgo.RequiredAcks(kgo.AllISRAcks()))
+	if err != nil {
+		return fmt.Errorf("create Kafka client: %w", err)
+	}
+	defer kafka.Close()
+	admin := kadm.NewClient(kafka)
+	runtime := liveRecoveryRuntime{
+		connectorDrainReader: connectorDrainReader{db: db, admin: admin, connect: &connect, name: opts.connectorName},
+		kafka:                kafka,
+		opts:                 opts,
+	}
+	return dependencies.runRecoveryFunc(ctx, opts, report, connect, runtime)
+}
+
+func runRecovery(ctx context.Context, opts runOptions, report *runReport, connect connectClient, runtime recoveryRuntime) error {
+	if err := setStage(opts, report, "prepare-target-table"); err != nil {
+		return err
+	}
+	phaseCtx, phaseCancel := context.WithTimeout(ctx, opts.phaseTimeout)
+	err := runtime.prepareTarget(phaseCtx)
+	phaseCancel()
+	if err != nil {
+		return err
+	}
+	if err := setStage(opts, report, "create-topic-and-connector"); err != nil {
+		return err
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	err = runtime.createTopic(phaseCtx)
+	phaseCancel()
+	if err != nil {
+		return err
+	}
+	config := sinkConfig(opts)
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	err = connect.createConnector(phaseCtx, opts.connectorName, config)
+	phaseCancel()
+	if err != nil {
+		return err
+	}
+	if err := setStage(opts, report, "wait-for-initial-connector"); err != nil {
+		return err
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	err = waitForConnector(phaseCtx, connect, opts.connectorName, "RUNNING")
+	phaseCancel()
+	if err != nil {
+		return err
+	}
+
+	baseline := []row{
+		{ID: 1, Name: "baseline-null", Note: nil, Counter: 1},
+		{ID: 2, Name: "雪-empty", Note: stringPointer(""), Counter: 2},
+		{ID: 3, Name: "before-update", Note: stringPointer("before"), Counter: 3},
+	}
+	if err := setStage(opts, report, "produce-baseline"); err != nil {
+		return err
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	baselineRecords, err := runtime.produceRows(phaseCtx, baseline)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("produce baseline: %w", err)
+	}
+	baselineWant := baseline
+	baselineOffset := lastOffset(baselineRecords) + 1
+	if err := setStage(opts, report, "baseline-drain"); err != nil {
+		return err
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	baselineReceipt, err := waitForDrain(phaseCtx, runtime, baselineWant, baselineOffset)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("baseline drain: %w", err)
+	}
+	if err := recordPhase(opts, report, phase{
+		Name: "baseline", Status: "passed", ExpectedOffset: baselineOffset,
+		RESTOffset: baselineReceipt.RESTOffset, KafkaGroupOffset: baselineReceipt.KafkaGroupOffset,
+		KafkaLogEndOffset: baselineReceipt.KafkaLogEndOffset, Produced: baselineRecords,
+		Rows: toRowReports(baselineWant), ConnectorState: "RUNNING",
+	}); err != nil {
+		return err
+	}
+
+	if err := setStage(opts, report, "stop-worker-and-produce-offline-records"); err != nil {
+		return err
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	err = runtime.killWorker(phaseCtx)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("SIGKILL owned Connect worker: %w", err)
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	currentOffset, ok, err := runtime.groupOffset(phaseCtx)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("read sink consumer group offset after worker crash: %w", err)
+	}
+	if !ok || currentOffset != baselineOffset {
+		return fmt.Errorf("crash checkpoint changed: got %d (present=%v), want baseline %d", currentOffset, ok, baselineOffset)
+	}
+	offline := []row{
+		{ID: 4, Name: "offline-insert", Note: stringPointer("while worker stopped"), Counter: 4},
+		{ID: 3, Name: "after-update", Note: stringPointer("更新"), Counter: 33},
+		{ID: 5, Name: "offline-null", Note: nil, Counter: 5},
+	}
+	if err := setStage(opts, report, "produce-offline-range"); err != nil {
+		return err
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	offlineRecords, err := runtime.produceRows(phaseCtx, offline[:1])
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("produce offline insert: %w", err)
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	updateRecords, err := runtime.produceRows(phaseCtx, offline[1:2])
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("produce offline update: %w", err)
+	}
+	offlineRecords = append(offlineRecords, updateRecords...)
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	tombstoneRecords, err := runtime.produceTombstone(phaseCtx, 1)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("produce keyed tombstone: %w", err)
+	}
+	offlineRecords = append(offlineRecords, tombstoneRecords...)
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	lastRecords, err := runtime.produceRows(phaseCtx, offline[2:])
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("produce final offline insert: %w", err)
+	}
+	offlineRecords = append(offlineRecords, lastRecords...)
+	finalOffset := lastOffset(offlineRecords) + 1
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	unchangedRows, err := runtime.rows(phaseCtx)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("read MatrixOne during worker outage: %w", err)
+	}
+	if !reflect.DeepEqual(unchangedRows, sortedRows(baselineWant)) {
+		return fmt.Errorf("MatrixOne changed while Connect was stopped: got %v, want baseline %v", unchangedRows, sortedRows(baselineWant))
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	got, ok, err := runtime.groupOffset(phaseCtx)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("read sink group offset while worker stopped: %w", err)
+	} else if !ok || got != baselineOffset {
+		return fmt.Errorf("sink checkpoint advanced while worker was stopped: got %d (present=%v), want %d", got, ok, baselineOffset)
+	}
+	allOffline := expectedFinalRows()
+	if err := validateUniqueRowIDs(allOffline); err != nil {
+		return fmt.Errorf("invalid expected final rows: %w", err)
+	}
+	if err := setStage(opts, report, "restart-worker-and-catch-up"); err != nil {
+		return err
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	err = runtime.startWorker(phaseCtx)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("restart same Connect worker: %w", err)
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	err = waitForConnector(phaseCtx, connect, opts.connectorName, "RUNNING")
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("connector recovery: %w", err)
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	catchUpReceipt, err := waitForDrain(phaseCtx, runtime, allOffline, finalOffset)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("post-crash catch-up: %w", err)
+	}
+	if err := recordPhase(opts, report, phase{
+		Name: "worker-crash-catch-up", Status: "passed", ExpectedOffset: finalOffset,
+		RESTOffset: catchUpReceipt.RESTOffset, KafkaGroupOffset: catchUpReceipt.KafkaGroupOffset,
+		KafkaLogEndOffset: catchUpReceipt.KafkaLogEndOffset,
+		Produced:          offlineRecords, Rows: toRowReports(allOffline), ConnectorState: "RUNNING", WorkerRestarted: true,
+		Detail: "same broker volume, Connect worker container, connector name/config, and sink consumer group; no connector re-registration or offset reset",
+	}); err != nil {
+		return err
+	}
+
+	if err := setStage(opts, report, "stop-connector-for-replay"); err != nil {
+		return err
+	}
+	if err := connect.stopConnector(ctx, opts.connectorName); err != nil {
+		return fmt.Errorf("stop connector for controlled replay: %w", err)
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	err = waitForConnector(phaseCtx, connect, opts.connectorName, "STOPPED")
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("wait for stopped connector before replay: %w", err)
+	}
+	if err := setStage(opts, report, "rewind-offset-for-replay"); err != nil {
+		return err
+	}
+	if err := connect.patchOffset(ctx, opts.connectorName, topicName, 0, offlineRecords[0].Offset); err != nil {
+		return fmt.Errorf("rewind owned sink offset for replay: %w", err)
+	}
+	replayedOffset, err := connect.restOffset(ctx, opts.connectorName, topicName, 0)
+	if err != nil {
+		return fmt.Errorf("verify replay offset reset: %w", err)
+	}
+	if replayedOffset != offlineRecords[0].Offset {
+		return fmt.Errorf("replay offset is %d, want %d", replayedOffset, offlineRecords[0].Offset)
+	}
+	if err := setStage(opts, report, "resume-connector-for-replay"); err != nil {
+		return err
+	}
+	if err := connect.resumeConnector(ctx, opts.connectorName); err != nil {
+		return fmt.Errorf("resume connector after controlled replay: %w", err)
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	err = waitForConnector(phaseCtx, connect, opts.connectorName, "RUNNING")
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("wait for connector after controlled replay: %w", err)
+	}
+	if err := setStage(opts, report, "replay-drain"); err != nil {
+		return err
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	replayReceipt, err := waitForDrain(phaseCtx, runtime, allOffline, finalOffset)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("controlled replay: %w", err)
+	}
+	if err := recordPhase(opts, report, phase{
+		Name: "controlled-replay", Status: "passed", ExpectedOffset: finalOffset,
+		RESTOffset: replayReceipt.RESTOffset, KafkaGroupOffset: replayReceipt.KafkaGroupOffset,
+		KafkaLogEndOffset: replayReceipt.KafkaLogEndOffset, Rows: toRowReports(allOffline),
+		ConnectorState: "RUNNING", ControlledReplay: true,
+		Detail: "connector stopped before the official offsets PATCH; replayed the owned offline range; exact destination rows remained unchanged",
+	}); err != nil {
+		return err
+	}
+
+	if err := setStage(opts, report, "empty-work-worker-restart"); err != nil {
+		return err
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	err = runtime.killWorker(phaseCtx)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("empty-work restart: stop owned Connect worker: %w", err)
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	err = runtime.startWorker(phaseCtx)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("empty-work restart: start same Connect worker: %w", err)
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	err = waitForConnector(phaseCtx, connect, opts.connectorName, "RUNNING")
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("empty-work restart: connector did not recover: %w", err)
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	stableRows, err := runtime.rows(phaseCtx)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("empty-work restart: read final rows: %w", err)
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	stableGroup, ok, err := runtime.groupOffset(phaseCtx)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("empty-work restart: read sink offset: %w", err)
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	stableREST, err := connect.restOffset(phaseCtx, opts.connectorName, topicName, 0)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("empty-work restart: read Connect REST offset: %w", err)
+	}
+	phaseCtx, phaseCancel = context.WithTimeout(ctx, opts.phaseTimeout)
+	stableLogEnd, endFound, err := runtime.logEndOffset(phaseCtx)
+	phaseCancel()
+	if err != nil {
+		return fmt.Errorf("empty-work restart: read topic log end: %w", err)
+	}
+	if !ok || stableREST != finalOffset || stableGroup != finalOffset || !endFound || stableLogEnd != finalOffset || !reflect.DeepEqual(stableRows, sortedRows(allOffline)) {
+		return fmt.Errorf("empty-work restart changed state: rows=%v REST offset=%d, group offset=%d (present=%v), log-end=%d (present=%v), want rows=%v offsets=%d", stableRows, stableREST, stableGroup, ok, stableLogEnd, endFound, sortedRows(allOffline), finalOffset)
+	}
+	if err := recordPhase(opts, report, phase{
+		Name: "empty-work-restart", Status: "passed", ExpectedOffset: finalOffset,
+		RESTOffset: stableREST, KafkaGroupOffset: stableGroup, KafkaLogEndOffset: stableLogEnd,
+		Rows: toRowReports(allOffline), ConnectorState: "RUNNING", WorkerRestarted: true,
+	}); err != nil {
+		return err
+	}
+	if err := setStage(opts, report, "complete"); err != nil {
+		return err
+	}
+	return nil
+}
+
+func sinkConfig(opts runOptions) map[string]string {
+	return map[string]string{
+		"connector.class":      connectorClass,
+		"tasks.max":            "1",
+		"connection.url":       fmt.Sprintf("jdbc:mysql://host.docker.internal:%s/%s?useSSL=false&serverTimezone=UTC&useServerPrepStmts=true&cachePrepStmts=true&rewriteBatchedStatements=false", portFromDSN(opts.dsn), databaseName),
+		"connection.username":  "root",
+		"connection.password":  "111",
+		"insert.mode":          "upsert",
+		"delete.enabled":       "true",
+		"primary.key.mode":     "record_key",
+		"schema.evolution":     "none",
+		"use.time.zone":        "UTC",
+		"topics":               topicName,
+		"table.name.format":    tableName,
+		"flush.max.retries":    "3",
+		"flush.retry.delay.ms": "500",
+		"batch.size":           "1",
+	}
+}
+
+func portFromDSN(dsn string) string {
+	start := strings.LastIndex(dsn, "tcp(")
+	if start < 0 {
+		return "6001"
+	}
+	start += len("tcp(")
+	end := strings.Index(dsn[start:], ")")
+	if end < 0 {
+		return "6001"
+	}
+	address := dsn[start : start+end]
+	_, port, err := net.SplitHostPort(address)
+	if err == nil && port != "" {
+		return port
+	}
+	return "6001"
+}
+
+type databasePinger interface {
+	PingContext(context.Context) error
+}
+
+func waitForDatabase(ctx context.Context, db databasePinger) error {
+	var lastPingErr error
+	err := poll(ctx, "MatrixOne SQL readiness", time.Second, func(ctx context.Context) (bool, error) {
+		if err := db.PingContext(ctx); err != nil {
+			lastPingErr = err
+			return false, nil
+		}
+		return true, nil
+	})
+	if err != nil && lastPingErr != nil {
+		return fmt.Errorf("%w (last ping: %v)", err, lastPingErr)
+	}
+	return err
+}
+
+func prepareTarget(ctx context.Context, db *sql.DB) error {
+	for _, statement := range []string{
+		"CREATE DATABASE IF NOT EXISTS " + databaseName,
+		"DROP TABLE IF EXISTS " + databaseName + "." + tableName,
+		"CREATE TABLE " + databaseName + "." + tableName + " (id BIGINT NOT NULL PRIMARY KEY, name VARCHAR(128) NOT NULL, note VARCHAR(255), counter BIGINT NOT NULL)",
+	} {
+		if _, err := db.ExecContext(ctx, statement); err != nil {
+			return fmt.Errorf("prepare run-owned target (%s): %w", statement, err)
+		}
+	}
+	return nil
+}
+
+func createTopic(ctx context.Context, admin *kadm.Client) error {
+	resp, err := admin.CreateTopic(ctx, 1, 1, nil, topicName)
+	if err != nil && !strings.Contains(err.Error(), "TOPIC_ALREADY_EXISTS") {
+		return fmt.Errorf("create one-partition fixture topic: %w", err)
+	}
+	if err == nil && resp.Err != nil && !strings.Contains(resp.Err.Error(), "TOPIC_ALREADY_EXISTS") {
+		return fmt.Errorf("create fixture topic: %w", resp.Err)
+	}
+	return nil
+}
+
+func produceRows(ctx context.Context, client *kgo.Client, rows []row) ([]producedRecord, error) {
+	records := make([]*kgo.Record, 0, len(rows))
+	for _, value := range rows {
+		key, err := encodeKey(value.ID)
+		if err != nil {
+			return nil, err
+		}
+		payload, err := encodeValue(value)
+		if err != nil {
+			return nil, err
+		}
+		records = append(records, &kgo.Record{Topic: topicName, Key: key, Value: payload})
+	}
+	return produce(ctx, client, records)
+}
+
+func produceTombstone(ctx context.Context, client *kgo.Client, id int64) ([]producedRecord, error) {
+	key, err := encodeKey(id)
+	if err != nil {
+		return nil, err
+	}
+	return produce(ctx, client, []*kgo.Record{{Topic: topicName, Key: key}})
+}
+
+func produce(ctx context.Context, client *kgo.Client, records []*kgo.Record) ([]producedRecord, error) {
+	results := client.ProduceSync(ctx, records...)
+	out := make([]producedRecord, 0, len(results))
+	for i, result := range results {
+		if result.Err != nil {
+			return nil, fmt.Errorf("produce record %d: %w", i, result.Err)
+		}
+		id, err := decodeKeyID(records[i].Key)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, producedRecord{Partition: result.Record.Partition, Offset: result.Record.Offset, ID: id, Tombstone: result.Record.Value == nil})
+	}
+	return out, nil
+}
+
+func encodeKey(id int64) ([]byte, error) {
+	return json.Marshal(map[string]any{
+		"schema":  map[string]any{"type": "struct", "name": "mo.issue29848.Key", "optional": false, "fields": []map[string]any{{"type": "int64", "optional": false, "field": "id"}}},
+		"payload": map[string]any{"id": id},
+	})
+}
+
+func decodeKeyID(raw []byte) (int64, error) {
+	var envelope struct {
+		Payload struct {
+			ID int64 `json:"id"`
+		} `json:"payload"`
+	}
+	if err := json.Unmarshal(raw, &envelope); err != nil {
+		return 0, err
+	}
+	return envelope.Payload.ID, nil
+}
+
+func encodeValue(value row) ([]byte, error) {
+	fields := []map[string]any{
+		{"type": "int64", "optional": false, "field": "id"},
+		{"type": "string", "optional": false, "field": "name"},
+		{"type": "string", "optional": true, "field": "note"},
+		{"type": "int64", "optional": false, "field": "counter"},
+	}
+	return json.Marshal(map[string]any{
+		"schema":  map[string]any{"type": "struct", "name": "mo.issue29848.Value", "optional": false, "fields": fields},
+		"payload": map[string]any{"id": value.ID, "name": value.Name, "note": value.Note, "counter": value.Counter},
+	})
+}
+
+func sortedRows(rows []row) []row {
+	out := append([]row(nil), rows...)
+	sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
+	return out
+}
+
+func expectedFinalRows() []row {
+	return []row{
+		{ID: 2, Name: "雪-empty", Note: stringPointer(""), Counter: 2},
+		{ID: 3, Name: "after-update", Note: stringPointer("更新"), Counter: 33},
+		{ID: 4, Name: "offline-insert", Note: stringPointer("while worker stopped"), Counter: 4},
+		{ID: 5, Name: "offline-null", Note: nil, Counter: 5},
+	}
+}
+
+func validateUniqueRowIDs(rows []row) error {
+	seen := make(map[int64]struct{}, len(rows))
+	for _, item := range rows {
+		if _, exists := seen[item.ID]; exists {
+			return fmt.Errorf("%w %d", errDuplicateRowID, item.ID)
+		}
+		seen[item.ID] = struct{}{}
+	}
+	return nil
+}
+
+func readRows(ctx context.Context, db *sql.DB) ([]row, error) {
+	rows, err := db.QueryContext(ctx, "SELECT id, name, note, counter FROM "+databaseName+"."+tableName+" ORDER BY id")
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []row
+	for rows.Next() {
+		var item row
+		var note sql.NullString
+		if err := rows.Scan(&item.ID, &item.Name, &note, &item.Counter); err != nil {
+			return nil, err
+		}
+		if note.Valid {
+			item.Note = &note.String
+		}
+		out = append(out, item)
+	}
+	return out, rows.Err()
+}
+
+func waitForConnect(ctx context.Context, connect connectClient) (connectRoot, error) {
+	var root connectRoot
+	var lastErr error
+	err := poll(ctx, "Kafka Connect REST readiness", time.Second, func(ctx context.Context) (bool, error) {
+		body, err := connect.request(ctx, http.MethodGet, "/", nil, http.StatusOK)
+		if err != nil {
+			lastErr = err
+			return false, nil
+		}
+		if err := json.Unmarshal(body, &root); err != nil {
+			return false, err
+		}
+		return root.Version != "", nil
+	})
+	if err != nil {
+		return root, fmt.Errorf("wait for Kafka Connect: %w (last REST error: %v)", err, lastErr)
+	}
+	return root, nil
+}
+
+func (c connectClient) plugins(ctx context.Context) ([]connectorPlugin, error) {
+	body, err := c.request(ctx, http.MethodGet, "/connector-plugins", nil, http.StatusOK)
+	if err != nil {
+		return nil, err
+	}
+	var plugins []connectorPlugin
+	if err := json.Unmarshal(body, &plugins); err != nil {
+		return nil, err
+	}
+	return plugins, nil
+}
+
+func (c connectClient) createConnector(ctx context.Context, name string, config map[string]string) error {
+	body, err := json.Marshal(map[string]any{"name": name, "config": config})
+	if err != nil {
+		return err
+	}
+	_, err = c.request(ctx, http.MethodPost, "/connectors", body, http.StatusCreated)
+	return err
+}
+
+func (c connectClient) status(ctx context.Context, name string) (connectorStatusResponse, error) {
+	body, err := c.request(ctx, http.MethodGet, "/connectors/"+url.PathEscape(name)+"/status", nil, http.StatusOK)
+	if err != nil {
+		return connectorStatusResponse{}, err
+	}
+	var status connectorStatusResponse
+	if err := json.Unmarshal(body, &status); err != nil {
+		return status, err
+	}
+	return status, nil
+}
+
+func (c connectClient) stopConnector(ctx context.Context, name string) error {
+	_, err := c.request(ctx, http.MethodPut, "/connectors/"+url.PathEscape(name)+"/stop", nil, http.StatusAccepted, http.StatusNoContent)
+	return err
+}
+
+func (c connectClient) resumeConnector(ctx context.Context, name string) error {
+	_, err := c.request(ctx, http.MethodPut, "/connectors/"+url.PathEscape(name)+"/resume", nil, http.StatusAccepted, http.StatusNoContent)
+	return err
+}
+
+func (c connectClient) restOffset(ctx context.Context, name, topic string, partition int32) (int64, error) {
+	body, err := c.request(ctx, http.MethodGet, "/connectors/"+url.PathEscape(name)+"/offsets", nil, http.StatusOK)
+	if err != nil {
+		return 0, err
+	}
+	var response connectorOffsets
+	if err := json.Unmarshal(body, &response); err != nil {
+		return 0, err
+	}
+	for _, item := range response.Offsets {
+		if item.Partition.KafkaTopic == topic && item.Partition.KafkaPartition == partition && item.Offset != nil {
+			return item.Offset.KafkaOffset, nil
+		}
+	}
+	return 0, fmt.Errorf("connector offsets omit %s[%d]", topic, partition)
+}
+
+func (c connectClient) patchOffset(ctx context.Context, name, topic string, partition int32, offset int64) error {
+	body, err := json.Marshal(connectorOffsets{Offsets: []connectorOffset{{
+		Partition: offsetPartition{KafkaTopic: topic, KafkaPartition: partition},
+		Offset:    &offsetValue{KafkaOffset: offset},
+	}}})
+	if err != nil {
+		return err
+	}
+	_, err = c.request(ctx, http.MethodPatch, "/connectors/"+url.PathEscape(name)+"/offsets", body, http.StatusAccepted, http.StatusOK, http.StatusNoContent)
+	return err
+}
+
+func (c connectClient) request(ctx context.Context, method, path string, body []byte, expected ...int) ([]byte, error) {
+	var reader io.Reader
+	if body != nil {
+		reader = bytes.NewReader(body)
+	}
+	req, err := http.NewRequestWithContext(ctx, method, c.base+path, reader)
+	if err != nil {
+		return nil, err
+	}
+	if body != nil {
+		req.Header.Set("Content-Type", "application/json")
+	}
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	data, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
+	if err != nil {
+		return nil, err
+	}
+	for _, status := range expected {
+		if resp.StatusCode == status {
+			return data, nil
+		}
+	}
+	return nil, fmt.Errorf("connect REST %s %s returned HTTP %d: %s", method, path, resp.StatusCode, strings.TrimSpace(string(data)))
+}
+
+func waitForConnector(ctx context.Context, connect connectClient, name, want string) error {
+	return poll(ctx, "connector state "+want, time.Second, func(ctx context.Context) (bool, error) {
+		status, err := connect.status(ctx, name)
+		if err != nil {
+			return false, nil
+		}
+		if status.Connector.State != want {
+			return false, nil
+		}
+		for _, task := range status.Tasks {
+			if task.State != want && !(want == "STOPPED" && task.State == "STOPPED") {
+				return false, nil
+			}
+		}
+		if want == "RUNNING" && len(status.Tasks) != 1 {
+			return false, nil
+		}
+		return true, nil
+	})
+}
+
+func waitForDrain(ctx context.Context, reader drainStateReader, expected []row, expectedOffset int64) (drainReceipt, error) {
+	var receipt drainReceipt
+	err := poll(ctx, "target rows and checkpoint", time.Second, func(ctx context.Context) (bool, error) {
+		actual, err := reader.rows(ctx)
+		if err != nil {
+			return false, nil
+		}
+		if !reflect.DeepEqual(actual, sortedRows(expected)) {
+			return false, nil
+		}
+		restOffset, err := reader.restOffset(ctx)
+		if err != nil || restOffset != expectedOffset {
+			return false, nil
+		}
+		kafkaOffset, groupFound, err := reader.groupOffset(ctx)
+		if err != nil || !groupFound || kafkaOffset != expectedOffset {
+			return false, nil
+		}
+		logEndOffset, endFound, err := reader.logEndOffset(ctx)
+		if err != nil || !endFound || logEndOffset != expectedOffset {
+			return false, nil
+		}
+		receipt = drainReceipt{
+			RESTOffset:        restOffset,
+			KafkaGroupOffset:  kafkaOffset,
+			KafkaLogEndOffset: logEndOffset,
+		}
+		return true, nil
+	})
+	return receipt, err
+}
+
+func groupOffset(ctx context.Context, admin *kadm.Client, group string) (int64, bool, error) {
+	response, err := admin.FetchOffsets(ctx, group)
+	if err != nil {
+		if strings.Contains(err.Error(), "GROUP_ID_NOT_FOUND") {
+			return 0, false, nil
+		}
+		return 0, false, err
+	}
+	offset, ok := response.Lookup(topicName, 0)
+	if !ok || offset.Err != nil {
+		return 0, false, offset.Err
+	}
+	return offset.At, true, nil
+}
+
+func topicLogEndOffset(ctx context.Context, admin *kadm.Client) (int64, bool, error) {
+	offsets, err := admin.ListEndOffsets(ctx, topicName)
+	if err != nil {
+		return 0, false, err
+	}
+	offset, ok := offsets.Lookup(topicName, 0)
+	if !ok {
+		return 0, false, nil
+	}
+	if offset.Err != nil {
+		return 0, false, offset.Err
+	}
+	return offset.Offset, true, nil
+}
+
+func killOwnedConnect(ctx context.Context, opts runOptions) error {
+	containerID, err := composeOutput(ctx, opts, "ps", "-q", "connect")
+	if err != nil {
+		return err
+	}
+	containerID = strings.TrimSpace(containerID)
+	if containerID == "" {
+		return errors.New("owned Connect container was not found in the run's Compose project")
+	}
+	if _, err := exec.CommandContext(ctx, "docker", "kill", "--signal", "KILL", containerID).CombinedOutput(); err != nil {
+		return fmt.Errorf("kill owned container %s: %w", containerID, err)
+	}
+	return poll(ctx, "owned Connect worker exit", 250*time.Millisecond, func(ctx context.Context) (bool, error) {
+		out, err := exec.CommandContext(ctx, "docker", "inspect", "--format", "{{.State.Status}}", containerID).CombinedOutput()
+		if err != nil {
+			return false, nil
+		}
+		return strings.TrimSpace(string(out)) == "exited", nil
+	})
+}
+
+func startOwnedConnect(ctx context.Context, opts runOptions) error {
+	_, err := composeOutput(ctx, opts, "start", "connect")
+	return err
+}
+
+func composeOutput(ctx context.Context, opts runOptions, args ...string) (string, error) {
+	composeBase, err := filepath.Abs(opts.composeBase)
+	if err != nil {
+		return "", fmt.Errorf("resolve base Compose path %q: %w", opts.composeBase, err)
+	}
+	composeConnect, err := filepath.Abs(opts.composeConnect)
+	if err != nil {
+		return "", fmt.Errorf("resolve Connect Compose path %q: %w", opts.composeConnect, err)
+	}
+	commandArgs := make([]string, 0, 7+len(args))
+	commandArgs = append(commandArgs, "compose", "-p", opts.composeProject, "-f", composeBase, "-f", composeConnect)
+	commandArgs = append(commandArgs, args...)
+	cmd := exec.CommandContext(ctx, "docker", commandArgs...)
+	cmd.Dir = filepath.Dir(composeBase)
+	output, err := cmd.CombinedOutput()
+	if err != nil {
+		return string(output), fmt.Errorf("docker %s: %w: %s", strings.Join(commandArgs, " "), err, strings.TrimSpace(string(output)))
+	}
+	return string(output), nil
+}
+
+func lastOffset(records []producedRecord) int64 {
+	last := int64(-1)
+	for _, record := range records {
+		if record.Offset > last {
+			last = record.Offset
+		}
+	}
+	return last
+}
+
+func toRowReports(rows []row) []rowReport {
+	sorted := sortedRows(rows)
+	out := make([]rowReport, 0, len(sorted))
+	for _, item := range sorted {
+		out = append(out, rowReport(item))
+	}
+	return out
+}
+
+func stringPointer(value string) *string { return &value }
+
+func poll(ctx context.Context, description string, interval time.Duration, check func(context.Context) (bool, error)) error {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		ok, err := check(ctx)
+		if err != nil {
+			return fmt.Errorf("%s: %w", description, err)
+		}
+		if ok {
+			return nil
+		}
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("%s timed out: %w", description, ctx.Err())
+		case <-ticker.C:
+		}
+	}
+}
+
+func setStage(opts runOptions, report *runReport, name string) error {
+	report.CurrentStage = name
+	fmt.Fprintf(os.Stdout, "Kafka Connect recovery stage: %s\n", name)
+	return writeReport(opts.reportDir, *report)
+}
+
+func recordPhase(opts runOptions, report *runReport, completed phase) error {
+	report.Phases = append(report.Phases, completed)
+	report.CurrentStage = completed.Name + "-passed"
+	fmt.Fprintf(os.Stdout, "Kafka Connect recovery phase passed: %s\n", completed.Name)
+	return writeReport(opts.reportDir, *report)
+}
+
+func writeReport(dir string, report runReport) error {
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return err
+	}
+	data, err := json.MarshalIndent(report, "", "  ")
+	if err != nil {
+		return err
+	}
+	tmp, err := os.CreateTemp(dir, ".report-*.tmp")
+	if err != nil {
+		return err
+	}
+	tmpPath := tmp.Name()
+	defer os.Remove(tmpPath)
+	if err := tmp.Chmod(0o644); err != nil {
+		tmp.Close()
+		return err
+	}
+	if _, err := tmp.Write(append(data, '\n')); err != nil {
+		tmp.Close()
+		return err
+	}
+	if err := tmp.Sync(); err != nil {
+		tmp.Close()
+		return err
+	}
+	if err := tmp.Close(); err != nil {
+		return err
+	}
+	if err := os.Rename(tmpPath, filepath.Join(dir, "report.json")); err != nil {
+		return err
+	}
+	return nil
+}
diff --git a/test/kafkaconnect/kafka_connect_e2e_local_test.go b/test/kafkaconnect/kafka_connect_e2e_local_test.go
new file mode 100644
index 0000000000000..a07ec85f8eda8
--- /dev/null
+++ b/test/kafkaconnect/kafka_connect_e2e_local_test.go
@@ -0,0 +1,1449 @@
+// Copyright 2026 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+
+package main
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"reflect"
+	"runtime"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/twmb/franz-go/pkg/kadm"
+	"github.com/twmb/franz-go/pkg/kfake"
+	"github.com/twmb/franz-go/pkg/kgo"
+)
+
+type staticDrainStateReader struct {
+	rowsValue        []row
+	restOffsetValue  int64
+	groupOffsetValue int64
+	groupFound       bool
+	logEndValue      int64
+	logEndFound      bool
+}
+
+func (r staticDrainStateReader) rows(context.Context) ([]row, error) {
+	return r.rowsValue, nil
+}
+
+func (r staticDrainStateReader) restOffset(context.Context) (int64, error) {
+	return r.restOffsetValue, nil
+}
+
+func (r staticDrainStateReader) groupOffset(context.Context) (int64, bool, error) {
+	return r.groupOffsetValue, r.groupFound, nil
+}
+
+func (r staticDrainStateReader) logEndOffset(context.Context) (int64, bool, error) {
+	return r.logEndValue, r.logEndFound, nil
+}
+
+func TestWaitForDrainReturnsEveryVerifiedOffset(t *testing.T) {
+	rows := expectedFinalRows()
+	reader := staticDrainStateReader{
+		rowsValue:        sortedRows(rows),
+		restOffsetValue:  7,
+		groupOffsetValue: 7,
+		groupFound:       true,
+		logEndValue:      7,
+		logEndFound:      true,
+	}
+	receipt, err := waitForDrain(t.Context(), reader, rows, 7)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if want := (drainReceipt{RESTOffset: 7, KafkaGroupOffset: 7, KafkaLogEndOffset: 7}); !reflect.DeepEqual(receipt, want) {
+		t.Fatalf("drain receipt = %+v, want %+v", receipt, want)
+	}
+}
+
+func TestWaitForDrainRejectsGroupOffsetBehindSinkAndLogEnd(t *testing.T) {
+	reader := staticDrainStateReader{
+		rowsValue:        sortedRows(expectedFinalRows()),
+		restOffsetValue:  7,
+		groupOffsetValue: 6,
+		groupFound:       true,
+		logEndValue:      7,
+		logEndFound:      true,
+	}
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Millisecond)
+	defer cancel()
+	if _, err := waitForDrain(ctx, reader, expectedFinalRows(), 7); err == nil {
+		t.Fatal("waitForDrain accepted a lagging Kafka consumer group")
+	}
+}
+
+func TestWaitForDrainRejectsRESTOffsetBehindSinkAndLogEnd(t *testing.T) {
+	reader := staticDrainStateReader{
+		rowsValue:        sortedRows(expectedFinalRows()),
+		restOffsetValue:  6,
+		groupOffsetValue: 7,
+		groupFound:       true,
+		logEndValue:      7,
+		logEndFound:      true,
+	}
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Millisecond)
+	defer cancel()
+	if _, err := waitForDrain(ctx, reader, expectedFinalRows(), 7); err == nil {
+		t.Fatal("waitForDrain accepted a lagging Connect REST offset")
+	}
+}
+
+func TestPhaseReportIncludesZeroOffsets(t *testing.T) {
+	data, err := json.Marshal(phase{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var fields map[string]json.RawMessage
+	if err := json.Unmarshal(data, &fields); err != nil {
+		t.Fatal(err)
+	}
+	for _, name := range []string{"expected_offset", "rest_offset", "kafka_group_offset", "kafka_log_end_offset"} {
+		if _, exists := fields[name]; !exists {
+			t.Errorf("phase receipt omitted %s at offset zero: %s", name, data)
+		}
+	}
+}
+
+func TestRunResultMarksFailedReport(t *testing.T) {
+	report := runReport{Status: "running"}
+	setRunResult(&report, errors.New("sink offset did not reach the expected value"))
+	data, err := json.Marshal(report)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var got struct {
+		Status string `json:"status"`
+		Error  string `json:"error"`
+	}
+	if err := json.Unmarshal(data, &got); err != nil {
+		t.Fatal(err)
+	}
+	if got.Status != "failed" || got.Error != "sink offset did not reach the expected value" {
+		t.Fatalf("failed run report = %+v, want failed status and error", got)
+	}
+	setRunResult(&report, nil)
+	if report.Status != "passed" || report.Error != "sink offset did not reach the expected value" {
+		t.Fatalf("successful run result = %+v, want passed status", report)
+	}
+}
+
+func TestKafkaConnectRunnerFailsWhenOnlyPortBlockIsLeased(t *testing.T) {
+	allocatePorts := kafkaConnectRunnerFunction(t, "allocate_ports", "collect_artifacts()")
+	script := strings.Join([]string{
+		"set -euo pipefail",
+		"PORT_BLOCK_WIDTH=80",
+		"PORT_BLOCK_BASE=''",
+		"MO_PORT='' STATUS_PORT='' KAFKA_PORT='' CONNECT_PORT=''",
+		"die() { echo \"$*\" >&2; exit 1; }",
+		"log() { :; }",
+		"acquire_port_lease() { return 1; }",
+		"mo_port_block_candidates() { echo 36000; }",
+		"mo_port_block_is_available() { return 0; }",
+		"mo_port_block_release_lease() { :; }",
+		allocatePorts,
+		"allocate_ports",
+	}, "\n")
+	cmd := exec.Command("bash", "-c", script)
+	output, runErr := cmd.CombinedOutput()
+	if runErr == nil {
+		t.Fatalf("runner succeeded despite the only port block being leased:\n%s", output)
+	}
+	if !strings.Contains(string(output), "could not find an available 80-port range") {
+		t.Fatalf("runner did not report exhausted port candidates (error %v):\n%s", runErr, output)
+	}
+}
+
+func TestKafkaConnectRunnerMarksCleanupFailureInReport(t *testing.T) {
+	cleanup := kafkaConnectRunnerFunction(t, "cleanup", "run_e2e()")
+	root := t.TempDir()
+	reportDir := filepath.Join(root, "reports")
+	if err := os.Mkdir(reportDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	tmpDir := filepath.Join(root, "mo-kafka-connect-e2e.test")
+	if err := os.Mkdir(tmpDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(reportDir, "report.json"), []byte(`{"status":"passed"}`), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	script := strings.Join([]string{
+		"set -u",
+		"CLEANUP_ERRORS=()",
+		"log() { :; }",
+		"record_cleanup_error() { CLEANUP_ERRORS+=(\"$1\"); }",
+		"mo_e2e_stop_service() { return 0; }",
+		"collect_artifacts() { return 0; }",
+		"compose() { return 19; }",
+		"mo_port_block_release_lease() { return 0; }",
+		cleanup,
+		"cleanup",
+	}, "\n")
+	cmd := exec.Command("bash", "-c", script)
+	cmd.Env = append(os.Environ(),
+		"REPORT_DIR="+reportDir,
+		"TMP_DIR="+tmpDir,
+		"MO_PID=",
+		"CURRENT_STAGE=run_recovery_driver",
+		"COMPOSE_PROJECT_NAME=mo-kc-test",
+		"SOURCE_SHA=test-revision",
+	)
+	output, runErr := cmd.CombinedOutput()
+	if runErr == nil {
+		t.Fatalf("cleanup returned success when Compose cleanup failed:\n%s", output)
+	}
+	var report struct {
+		Status        string   `json:"status"`
+		CleanupStatus string   `json:"cleanup_status"`
+		CleanupErrors []string `json:"cleanup_errors"`
+		Error         string   `json:"error"`
+	}
+	data, err := os.ReadFile(filepath.Join(reportDir, "report.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := json.Unmarshal(data, &report); err != nil {
+		t.Fatal(err)
+	}
+	if report.Status != "failed" || report.CleanupStatus != "failed" || len(report.CleanupErrors) == 0 || !strings.Contains(report.Error, "docker compose down failed") {
+		t.Fatalf("cleanup failure report = %+v, want failed status and Compose cleanup error; runner output: %s", report, output)
+	}
+}
+
+func TestKafkaConnectRunnerFailsWhenArtifactRedactionFails(t *testing.T) {
+	collectArtifacts := kafkaConnectRunnerFunction(t, "collect_artifacts", "cleanup()")
+	root := t.TempDir()
+	reportDir := filepath.Join(root, "reports")
+	tmpDir := filepath.Join(root, "run")
+	if err := os.Mkdir(reportDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Mkdir(tmpDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	psCalled := filepath.Join(root, "ps-called")
+	script := strings.Join([]string{
+		"set -u",
+		"TMP_DIR=\"$TEST_TMP_DIR\"",
+		"REPORT_DIR=\"$TEST_REPORT_DIR\"",
+		"compose() { case \"$1\" in logs) return 0 ;; ps) : >\"$TEST_PS_CALLED\"; return 0 ;; esac; }",
+		"python3() { return 17; }",
+		collectArtifacts,
+		"collect_artifacts",
+	}, "\n")
+	cmd := exec.Command("bash", "-c", script)
+	cmd.Env = append(os.Environ(),
+		"TEST_TMP_DIR="+tmpDir,
+		"TEST_REPORT_DIR="+reportDir,
+		"TEST_PS_CALLED="+psCalled,
+	)
+	output, runErr := cmd.CombinedOutput()
+	if runErr == nil {
+		t.Fatalf("artifact collection succeeded when log redaction failed:\n%s", output)
+	}
+	if _, err := os.Stat(psCalled); !os.IsNotExist(err) {
+		t.Fatalf("Compose ps ran after log redaction failed; got stat error %v", err)
+	}
+}
+
+func TestKafkaConnectRunnerUsesFreshReportBeforePreflight(t *testing.T) {
+	_, testFile, _, ok := runtime.Caller(0)
+	if !ok {
+		t.Fatal("could not locate test source")
+	}
+	repoRoot := filepath.Clean(filepath.Join(filepath.Dir(testFile), "..", ".."))
+	reportDir := filepath.Join(t.TempDir(), "reports")
+	if err := os.Mkdir(reportDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	for name, contents := range map[string]string{
+		"report.json":        `{"status":"passed"}`,
+		"setup-failure.json": `{"status":"failed","stage":"previous-run"}`,
+		"keep-me.txt":        "unrelated evidence\n",
+	} {
+		if err := os.WriteFile(filepath.Join(reportDir, name), []byte(contents), 0o644); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// Keep docker out of PATH to fail in preflight before creating any owned
+	// service, while preserving the small set of tools needed to write evidence.
+	pathDir := t.TempDir()
+	for _, name := range []string{"dirname", "mkdir", "python3", "rm"} {
+		path, err := exec.LookPath(name)
+		if err != nil {
+			t.Fatalf("look up %s: %v", name, err)
+		}
+		if err := os.Symlink(path, filepath.Join(pathDir, name)); err != nil {
+			t.Fatalf("link %s: %v", name, err)
+		}
+	}
+
+	cmd := exec.Command("/bin/bash", filepath.Join(repoRoot, "optools", "kafka_connect_ci.bash"), "e2e-local")
+	for _, item := range os.Environ() {
+		if strings.HasPrefix(item, "PATH=") || strings.HasPrefix(item, "MO_KAFKA_CONNECT_REPORT_DIR=") {
+			continue
+		}
+		cmd.Env = append(cmd.Env, item)
+	}
+	cmd.Env = append(cmd.Env, "PATH="+pathDir, "MO_KAFKA_CONNECT_REPORT_DIR="+reportDir)
+	output, runErr := cmd.CombinedOutput()
+	if runErr == nil || !strings.Contains(string(output), "required command not found: docker") {
+		t.Fatalf("runner preflight error = %v, output:\n%s", runErr, output)
+	}
+	if _, err := os.Stat(filepath.Join(reportDir, "report.json")); !os.IsNotExist(err) {
+		t.Fatalf("stale report.json remains after failed run; stat error = %v", err)
+	}
+	var failure struct {
+		Status    string `json:"status"`
+		Stage     string `json:"stage"`
+		SourceSHA string `json:"source_sha"`
+		ExitCode  int    `json:"exit_status"`
+	}
+	failureData, err := os.ReadFile(filepath.Join(reportDir, "setup-failure.json"))
+	if err != nil {
+		t.Fatalf("read current setup failure: %v; runner output:\n%s", err, output)
+	}
+	if err := json.Unmarshal(failureData, &failure); err != nil {
+		t.Fatalf("decode current setup failure: %v", err)
+	}
+	if failure.Status != "failed" || failure.Stage != "preflight" || failure.ExitCode == 0 {
+		t.Fatalf("setup-failure.json = %+v, want current failed preflight evidence", failure)
+	}
+	if got, err := os.ReadFile(filepath.Join(reportDir, "keep-me.txt")); err != nil || string(got) != "unrelated evidence\n" {
+		t.Fatalf("unrelated report artifact = %q, %v; want it preserved", got, err)
+	}
+}
+
+func TestWaitForDatabaseRetriesTransientPingFailure(t *testing.T) {
+	pinger := &sequenceDatabasePinger{}
+	if err := waitForDatabase(t.Context(), pinger); err != nil {
+		t.Fatalf("waitForDatabase() returned error after a transient PingContext failure: %v", err)
+	}
+	if pinger.calls != 2 {
+		t.Fatalf("PingContext calls = %d, want first failed ping followed by successful retry", pinger.calls)
+	}
+}
+
+type sequenceDatabasePinger struct {
+	calls int
+}
+
+func (p *sequenceDatabasePinger) PingContext(context.Context) error {
+	p.calls++
+	if p.calls == 1 {
+		return errors.New("transient connection refusal")
+	}
+	return nil
+}
+
+func TestSinkConfigUsesDebeziumJdbcBatchSize(t *testing.T) {
+	config := sinkConfig(runOptions{dsn: "root:111@tcp(127.0.0.1:6001)/"})
+	if got := config["batch.size"]; got != "1" {
+		t.Fatalf("batch.size = %q, want 1", got)
+	}
+	if _, exists := config["max.batch.size"]; exists {
+		t.Fatal("sink config contains unsupported max.batch.size property")
+	}
+}
+
+func kafkaConnectRunnerFunction(t *testing.T, name, next string) string {
+	t.Helper()
+	_, testFile, _, ok := runtime.Caller(0)
+	if !ok {
+		t.Fatal("could not locate test source")
+	}
+	repoRoot := filepath.Clean(filepath.Join(filepath.Dir(testFile), "..", ".."))
+	runner, err := os.ReadFile(filepath.Join(repoRoot, "optools", "kafka_connect_ci.bash"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	startToken := name + "() {"
+	start := strings.Index(string(runner), startToken)
+	if start < 0 {
+		t.Fatalf("could not find %s in Kafka Connect runner", startToken)
+	}
+	endToken := "\n}\n\n" + next
+	end := strings.Index(string(runner)[start:], endToken)
+	if end < 0 {
+		t.Fatalf("could not find end marker %q for %s", next, name)
+	}
+	return string(runner[start : start+end+2])
+}
+
+func TestExpectedFinalRowsHaveUniqueIDsAndCoverRecoveryContract(t *testing.T) {
+	rows := expectedFinalRows()
+	if err := validateUniqueRowIDs(rows); err != nil {
+		t.Fatalf("final recovery fixture must contain each primary key once: %v", err)
+	}
+	if len(rows) != 4 {
+		t.Fatalf("final row count = %d, want 4", len(rows))
+	}
+	byID := make(map[int64]row, len(rows))
+	for _, item := range rows {
+		byID[item.ID] = item
+	}
+	if _, exists := byID[1]; exists {
+		t.Fatal("final recovery fixture must omit the tombstoned row with ID 1")
+	}
+	if got := byID[2].Note; got == nil || *got != "" {
+		t.Fatalf("ID 2 note = %v, want a non-NULL empty string", got)
+	}
+	if got := byID[3]; got.Name != "after-update" || got.Counter != 33 || got.Note == nil || *got.Note != "更新" {
+		t.Fatalf("ID 3 = %+v, want the updated row", got)
+	}
+	if got := byID[5].Note; got != nil {
+		t.Fatalf("ID 5 note = %q, want SQL NULL", *got)
+	}
+
+	duplicate := append(append([]row(nil), rows...), rows[0])
+	if err := validateUniqueRowIDs(duplicate); err == nil || !errors.Is(err, errDuplicateRowID) {
+		t.Fatalf("duplicate primary key error = %v, want duplicate-row error", err)
+	}
+}
+
+func TestJSONConverterEnvelopesPreserveKeyAndNullSemantics(t *testing.T) {
+	key, err := encodeKey(17)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var decodedKey struct {
+		Schema struct {
+			Fields []struct {
+				Field string `json:"field"`
+			} `json:"fields"`
+		} `json:"schema"`
+		Payload map[string]int64 `json:"payload"`
+	}
+	if err := json.Unmarshal(key, &decodedKey); err != nil {
+		t.Fatal(err)
+	}
+	if len(decodedKey.Schema.Fields) != 1 || decodedKey.Schema.Fields[0].Field != "id" || decodedKey.Payload["id"] != 17 {
+		t.Fatalf("unexpected Kafka key envelope: %s", key)
+	}
+
+	for _, test := range []struct {
+		name string
+		note *string
+		want any
+	}{
+		{name: "SQL NULL", note: nil, want: nil},
+		{name: "empty string", note: stringPointer(""), want: ""},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			value, err := encodeValue(row{ID: 17, Name: "fixture", Note: test.note, Counter: 1})
+			if err != nil {
+				t.Fatal(err)
+			}
+			var decoded struct {
+				Schema struct {
+					Fields []struct {
+						Field    string `json:"field"`
+						Optional bool   `json:"optional"`
+					} `json:"fields"`
+				} `json:"schema"`
+				Payload map[string]any `json:"payload"`
+			}
+			if err := json.Unmarshal(value, &decoded); err != nil {
+				t.Fatal(err)
+			}
+			if len(decoded.Schema.Fields) != 4 || decoded.Schema.Fields[2].Field != "note" || !decoded.Schema.Fields[2].Optional {
+				t.Fatalf("note must be an optional schema field: %s", value)
+			}
+			if decoded.Payload["note"] != test.want {
+				t.Fatalf("note payload = %#v, want %#v", decoded.Payload["note"], test.want)
+			}
+		})
+	}
+}
+
+type recoveryTestEvent struct {
+	offset      int64
+	value       *row
+	tombstoneID *int64
+}
+
+type recoveryTestRuntime struct {
+	rowsValue        []row
+	rowsByID         map[int64]row
+	events           []recoveryTestEvent
+	nextOffset       int64
+	groupOffsetValue int64
+	restOffsetValue  int64
+	groupFound       bool
+	workerStopped    bool
+	actions          []string
+	prepareErr       error
+	createTopicErr   error
+}
+
+func (r *recoveryTestRuntime) prepareTarget(context.Context) error {
+	r.actions = append(r.actions, "prepare")
+	if r.prepareErr != nil {
+		return r.prepareErr
+	}
+	r.rowsByID = make(map[int64]row)
+	r.rowsValue = nil
+	r.events = nil
+	r.nextOffset = 0
+	r.groupOffsetValue = 0
+	r.restOffsetValue = 0
+	r.groupFound = false
+	r.workerStopped = false
+	return nil
+}
+
+func (r *recoveryTestRuntime) createTopic(context.Context) error {
+	r.actions = append(r.actions, "topic")
+	return r.createTopicErr
+}
+
+func (r *recoveryTestRuntime) produceRows(_ context.Context, values []row) ([]producedRecord, error) {
+	records := make([]producedRecord, 0, len(values))
+	for _, value := range values {
+		copyValue := value
+		r.events = append(r.events, recoveryTestEvent{offset: r.nextOffset, value: &copyValue})
+		records = append(records, producedRecord{Partition: 0, Offset: r.nextOffset, ID: value.ID})
+		r.nextOffset++
+		if !r.workerStopped {
+			r.apply(r.events[len(r.events)-1])
+		}
+	}
+	if !r.workerStopped {
+		r.groupOffsetValue = r.nextOffset
+		r.restOffsetValue = r.nextOffset
+		r.groupFound = true
+	}
+	r.actions = append(r.actions, "produce")
+	return records, nil
+}
+
+func (r *recoveryTestRuntime) produceTombstone(_ context.Context, id int64) ([]producedRecord, error) {
+	idCopy := id
+	r.events = append(r.events, recoveryTestEvent{offset: r.nextOffset, tombstoneID: &idCopy})
+	record := producedRecord{Partition: 0, Offset: r.nextOffset, ID: id, Tombstone: true}
+	r.nextOffset++
+	if !r.workerStopped {
+		r.apply(r.events[len(r.events)-1])
+		r.groupOffsetValue = r.nextOffset
+		r.restOffsetValue = r.nextOffset
+		r.groupFound = true
+	}
+	r.actions = append(r.actions, "tombstone")
+	return []producedRecord{record}, nil
+}
+
+func (r *recoveryTestRuntime) killWorker(context.Context) error {
+	r.workerStopped = true
+	r.actions = append(r.actions, "kill")
+	return nil
+}
+
+func (r *recoveryTestRuntime) startWorker(context.Context) error {
+	r.workerStopped = false
+	r.rowsByID = make(map[int64]row)
+	for _, event := range r.events {
+		r.apply(event)
+	}
+	r.groupOffsetValue = r.nextOffset
+	r.restOffsetValue = r.nextOffset
+	r.groupFound = true
+	r.actions = append(r.actions, "start")
+	return nil
+}
+
+func (r *recoveryTestRuntime) rows(context.Context) ([]row, error) {
+	return append([]row(nil), r.rowsValue...), nil
+}
+
+func (r *recoveryTestRuntime) restOffset(context.Context) (int64, error) {
+	return r.restOffsetValue, nil
+}
+
+func (r *recoveryTestRuntime) groupOffset(context.Context) (int64, bool, error) {
+	return r.groupOffsetValue, r.groupFound, nil
+}
+
+func (r *recoveryTestRuntime) logEndOffset(context.Context) (int64, bool, error) {
+	return r.nextOffset, r.nextOffset > 0, nil
+}
+
+func (r *recoveryTestRuntime) apply(event recoveryTestEvent) {
+	if event.tombstoneID != nil {
+		delete(r.rowsByID, *event.tombstoneID)
+	} else if event.value != nil {
+		r.rowsByID[event.value.ID] = *event.value
+	}
+	values := make([]row, 0, len(r.rowsByID))
+	for _, value := range r.rowsByID {
+		values = append(values, value)
+	}
+	r.rowsValue = sortedRows(values)
+}
+
+type recoveryTestConnect struct {
+	runtime      *recoveryTestRuntime
+	state        string
+	patchErr     error
+	actions      []string
+	registers    int
+	patchStopped bool
+	stoppedSeen  bool
+}
+
+func (c *recoveryTestConnect) handler(w http.ResponseWriter, request *http.Request) {
+	writeJSON := func(value any) {
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(value)
+	}
+	switch {
+	case request.Method == http.MethodPost && request.URL.Path == "/connectors":
+		c.registers++
+		c.actions = append(c.actions, "register")
+		c.state = "RUNNING"
+		w.WriteHeader(http.StatusCreated)
+	case request.Method == http.MethodGet && strings.HasSuffix(request.URL.Path, "/status"):
+		statusState := c.state
+		if c.state == "STOPPING" {
+			c.state = "STOPPED"
+		} else if statusState == "STOPPED" {
+			c.stoppedSeen = true
+		}
+		writeJSON(connectorStatusResponse{
+			Connector: connectorStatus{Name: "fixture", State: statusState, Worker: "worker-0"},
+			Tasks:     []taskStatus{{ID: 0, State: statusState, Worker: "worker-0"}},
+		})
+	case request.Method == http.MethodPut && strings.HasSuffix(request.URL.Path, "/stop"):
+		c.actions = append(c.actions, "stop")
+		c.state = "STOPPING"
+		w.WriteHeader(http.StatusAccepted)
+	case request.Method == http.MethodPut && strings.HasSuffix(request.URL.Path, "/resume"):
+		c.actions = append(c.actions, "resume")
+		c.state = "RUNNING"
+		c.runtime.restOffsetValue = c.runtime.nextOffset
+		w.WriteHeader(http.StatusAccepted)
+	case request.Method == http.MethodPatch && strings.HasSuffix(request.URL.Path, "/offsets"):
+		c.actions = append(c.actions, "patch")
+		if c.state != "STOPPED" {
+			http.Error(w, "connector is not stopped", http.StatusConflict)
+			return
+		}
+		if c.patchErr != nil {
+			http.Error(w, c.patchErr.Error(), http.StatusInternalServerError)
+			return
+		}
+		var offsets connectorOffsets
+		if err := json.NewDecoder(request.Body).Decode(&offsets); err != nil || len(offsets.Offsets) != 1 || offsets.Offsets[0].Offset == nil {
+			http.Error(w, "malformed offsets", http.StatusBadRequest)
+			return
+		}
+		c.runtime.restOffsetValue = offsets.Offsets[0].Offset.KafkaOffset
+		c.patchStopped = true
+		w.WriteHeader(http.StatusAccepted)
+	case request.Method == http.MethodGet && strings.HasSuffix(request.URL.Path, "/offsets"):
+		writeJSON(connectorOffsets{Offsets: []connectorOffset{{
+			Partition: offsetPartition{KafkaTopic: topicName, KafkaPartition: 0},
+			Offset:    &offsetValue{KafkaOffset: c.runtime.restOffsetValue},
+		}}})
+	default:
+		http.NotFound(w, request)
+	}
+}
+
+func newRecoveryFixture(t *testing.T, patchErr error) (*recoveryTestRuntime, *recoveryTestConnect, connectClient, runOptions, *runReport, func()) {
+	t.Helper()
+	runtime := &recoveryTestRuntime{}
+	serverState := &recoveryTestConnect{runtime: runtime, patchErr: patchErr}
+	server := httptest.NewServer(http.HandlerFunc(serverState.handler))
+	opts := runOptions{connectorName: "fixture", reportDir: t.TempDir(), phaseTimeout: 3 * time.Second}
+	report := &runReport{SinkConsumerGroup: "connect-fixture", Phases: []phase{}}
+	connect := connectClient{base: server.URL, client: server.Client()}
+	return runtime, serverState, connect, opts, report, server.Close
+}
+
+func TestRunRecoveryExercisesSuccessfulRecoveryAndReplay(t *testing.T) {
+	runtime, server, connect, opts, report, closeServer := newRecoveryFixture(t, nil)
+	defer closeServer()
+	if err := runRecovery(t.Context(), opts, report, connect, runtime); err != nil {
+		t.Fatal(err)
+	}
+	if runtime.rowsValue == nil || !reflect.DeepEqual(runtime.rowsValue, sortedRows(expectedFinalRows())) {
+		t.Fatalf("final fake rows = %v, want %v", runtime.rowsValue, sortedRows(expectedFinalRows()))
+	}
+	if server.registers != 1 {
+		t.Fatalf("connector registrations = %d, want one registration across recovery", server.registers)
+	}
+	wantActions := []string{"register", "stop", "patch", "resume"}
+	for _, want := range wantActions {
+		found := false
+		for _, action := range server.actions {
+			if action == want {
+				found = true
+				break
+			}
+		}
+		if !found {
+			t.Fatalf("REST actions = %v, missing %q", server.actions, want)
+		}
+	}
+	patchIndex := -1
+	resumeIndex := -1
+	stopIndex := -1
+	for index, action := range server.actions {
+		switch action {
+		case "stop":
+			stopIndex = index
+		case "patch":
+			patchIndex = index
+		case "resume":
+			resumeIndex = index
+		}
+	}
+	if !(stopIndex >= 0 && stopIndex < patchIndex && patchIndex < resumeIndex) {
+		t.Fatalf("REST actions = %v, want STOPPED before PATCH before resume", server.actions)
+	}
+	if !server.patchStopped {
+		t.Fatalf("PATCH was accepted without a confirmed STOPPED status: %v", server.actions)
+	}
+	if !server.stoppedSeen {
+		t.Fatalf("PATCH proceeded without observing a STOPPED status response: %v", server.actions)
+	}
+	if len(report.Phases) != 4 || report.Phases[0].Name != "baseline" || report.Phases[1].Name != "worker-crash-catch-up" || report.Phases[2].Name != "controlled-replay" || report.Phases[3].Name != "empty-work-restart" {
+		t.Fatalf("recovery phases = %+v, want baseline, catch-up, replay, and empty restart", report.Phases)
+	}
+}
+
+func TestRunRecoveryStopsWhenPatchIsRejected(t *testing.T) {
+	patchErr := errors.New("offset patch rejected")
+	runtime, server, connect, opts, report, closeServer := newRecoveryFixture(t, patchErr)
+	defer closeServer()
+	if err := runRecovery(t.Context(), opts, report, connect, runtime); err == nil || !strings.Contains(err.Error(), "rewind owned sink offset") {
+		t.Fatalf("runRecovery error = %v, want offset rewind failure", err)
+	}
+	for _, action := range server.actions {
+		if action == "resume" {
+			t.Fatalf("REST resumed after rejected patch: %v", server.actions)
+		}
+	}
+}
+
+func TestRunRecoveryStopsBeforeConnectorWhenTargetPreparationFails(t *testing.T) {
+	runtime := &recoveryTestRuntime{prepareErr: errors.New("target unavailable")}
+	opts := runOptions{connectorName: "fixture", reportDir: t.TempDir(), phaseTimeout: time.Second}
+	report := &runReport{SinkConsumerGroup: "connect-fixture", Phases: []phase{}}
+	connect := connectClient{base: "http://127.0.0.1:1", client: &http.Client{}}
+	if err := runRecovery(t.Context(), opts, report, connect, runtime); err == nil || !strings.Contains(err.Error(), "target unavailable") {
+		t.Fatalf("runRecovery error = %v, want target preparation failure", err)
+	}
+	if len(runtime.actions) != 1 || runtime.actions[0] != "prepare" {
+		t.Fatalf("runtime actions = %v, want preparation to be the only action", runtime.actions)
+	}
+}
+
+func TestPrepareTargetAndReadRows(t *testing.T) {
+	t.Run("prepare succeeds", func(t *testing.T) {
+		db, mock, err := sqlmock.New()
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer db.Close()
+		for _, statement := range []string{
+			"CREATE DATABASE IF NOT EXISTS " + databaseName,
+			"DROP TABLE IF EXISTS " + databaseName + "." + tableName,
+			"CREATE TABLE " + databaseName + "." + tableName,
+		} {
+			mock.ExpectExec(statement).WillReturnResult(sqlmock.NewResult(0, 1))
+		}
+		if err := prepareTarget(t.Context(), db); err != nil {
+			t.Fatal(err)
+		}
+		if err := mock.ExpectationsWereMet(); err != nil {
+			t.Fatal(err)
+		}
+	})
+
+	t.Run("prepare stops on first failed statement", func(t *testing.T) {
+		db, mock, err := sqlmock.New()
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer db.Close()
+		mock.ExpectExec("CREATE DATABASE IF NOT EXISTS").WillReturnResult(sqlmock.NewResult(0, 1))
+		mock.ExpectExec("DROP TABLE IF EXISTS").WillReturnError(errors.New("table locked"))
+		if err := prepareTarget(t.Context(), db); err == nil || !strings.Contains(err.Error(), "table locked") {
+			t.Fatalf("prepareTarget error = %v, want table lock failure", err)
+		}
+		if err := mock.ExpectationsWereMet(); err != nil {
+			t.Fatal(err)
+		}
+	})
+
+	t.Run("read rows preserves null and empty values", func(t *testing.T) {
+		db, mock, err := sqlmock.New()
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer db.Close()
+		empty := ""
+		mock.ExpectQuery("SELECT id, name, note, counter").WillReturnRows(
+			sqlmock.NewRows([]string{"id", "name", "note", "counter"}).
+				AddRow(int64(1), "null-note", nil, int64(1)).
+				AddRow(int64(2), "empty-note", empty, int64(2)),
+		)
+		got, err := readRows(t.Context(), db)
+		if err != nil {
+			t.Fatal(err)
+		}
+		want := []row{{ID: 1, Name: "null-note", Counter: 1}, {ID: 2, Name: "empty-note", Note: &empty, Counter: 2}}
+		if !reflect.DeepEqual(got, want) {
+			t.Fatalf("readRows = %+v, want %+v", got, want)
+		}
+		if err := mock.ExpectationsWereMet(); err != nil {
+			t.Fatal(err)
+		}
+	})
+
+	t.Run("read rows reports query and scan failures", func(t *testing.T) {
+		db, mock, err := sqlmock.New()
+		if err != nil {
+			t.Fatal(err)
+		}
+		mock.ExpectQuery("SELECT id, name, note, counter").WillReturnError(errors.New("database offline"))
+		if _, err := readRows(t.Context(), db); err == nil || !strings.Contains(err.Error(), "database offline") {
+			t.Fatalf("readRows query error = %v", err)
+		}
+		mock.ExpectQuery("SELECT id, name, note, counter").WillReturnRows(
+			sqlmock.NewRows([]string{"id", "name", "note", "counter"}).AddRow("bad-id", "name", nil, int64(1)),
+		)
+		if _, err := readRows(t.Context(), db); err == nil {
+			t.Fatal("readRows accepted a row that cannot be scanned")
+		}
+		_ = db.Close()
+	})
+}
+
+func TestConnectClientReadinessAndConnectorOperations(t *testing.T) {
+	var receivedConfig map[string]any
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		switch {
+		case request.Method == http.MethodGet && request.URL.Path == "/":
+			_, _ = w.Write([]byte(`{"version":"3.8.1","kafka_cluster_id":"cluster-1"}`))
+		case request.Method == http.MethodGet && request.URL.Path == "/connector-plugins":
+			_, _ = w.Write([]byte(`[{"class":"io.debezium.connector.jdbc.JdbcSinkConnector","type":"sink","version":"2.6"}]`))
+		case request.Method == http.MethodPost && request.URL.Path == "/connectors":
+			if err := json.NewDecoder(request.Body).Decode(&receivedConfig); err != nil {
+				http.Error(w, err.Error(), http.StatusBadRequest)
+				return
+			}
+			w.WriteHeader(http.StatusCreated)
+		case request.Method == http.MethodGet && strings.HasSuffix(request.URL.Path, "/status"):
+			_, _ = w.Write([]byte(`{"connector":{"name":"fixture","state":"RUNNING","worker_id":"worker-0"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"worker-0"}]}`))
+		case request.Method == http.MethodPut && (strings.HasSuffix(request.URL.Path, "/stop") || strings.HasSuffix(request.URL.Path, "/resume")):
+			w.WriteHeader(http.StatusNoContent)
+		case request.Method == http.MethodGet && strings.HasSuffix(request.URL.Path, "/offsets"):
+			_, _ = w.Write([]byte(`{"offsets":[{"partition":{"kafka_topic":"mo_issue_29848_events","kafka_partition":0},"offset":{"kafka_offset":12}}]}`))
+		case request.Method == http.MethodPatch && strings.HasSuffix(request.URL.Path, "/offsets"):
+			w.WriteHeader(http.StatusAccepted)
+		default:
+			http.NotFound(w, request)
+		}
+	}))
+	defer server.Close()
+	connect := connectClient{base: server.URL, client: server.Client()}
+	root, err := waitForConnect(t.Context(), connect)
+	if err != nil || root.Version != "3.8.1" || root.KafkaClusterID != "cluster-1" {
+		t.Fatalf("waitForConnect = %+v, %v", root, err)
+	}
+	plugins, err := connect.plugins(t.Context())
+	if err != nil || len(plugins) != 1 || plugins[0].Version != "2.6" {
+		t.Fatalf("plugins = %+v, %v", plugins, err)
+	}
+	if err := connect.createConnector(t.Context(), "fixture", map[string]string{"topics": topicName}); err != nil {
+		t.Fatal(err)
+	}
+	if receivedConfig["name"] != "fixture" {
+		t.Fatalf("connector request = %#v, want connector name", receivedConfig)
+	}
+	status, err := connect.status(t.Context(), "fixture")
+	if err != nil || status.Connector.State != "RUNNING" || len(status.Tasks) != 1 {
+		t.Fatalf("status = %+v, %v", status, err)
+	}
+	if err := connect.stopConnector(t.Context(), "fixture"); err != nil {
+		t.Fatal(err)
+	}
+	if err := connect.resumeConnector(t.Context(), "fixture"); err != nil {
+		t.Fatal(err)
+	}
+	offset, err := connect.restOffset(t.Context(), "fixture", topicName, 0)
+	if err != nil || offset != 12 {
+		t.Fatalf("restOffset = %d, %v", offset, err)
+	}
+	if err := connect.patchOffset(t.Context(), "fixture", topicName, 0, 7); err != nil {
+		t.Fatal(err)
+	}
+
+	badJSON := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
+		if request.URL.Path == "/connector-plugins" || strings.HasSuffix(request.URL.Path, "/status") {
+			_, _ = w.Write([]byte("{"))
+			return
+		}
+		_, _ = w.Write([]byte(`{}`))
+	}))
+	defer badJSON.Close()
+	badConnect := connectClient{base: badJSON.URL, client: badJSON.Client()}
+	if _, err := badConnect.plugins(t.Context()); err == nil {
+		t.Fatal("plugins accepted malformed JSON")
+	}
+	if _, err := badConnect.status(t.Context(), "fixture"); err == nil {
+		t.Fatal("status accepted malformed JSON")
+	}
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	if _, err := waitForConnect(ctx, badConnect); err == nil {
+		t.Fatal("waitForConnect accepted a canceled readiness context")
+	}
+	if _, err := (connectClient{base: "http://[", client: http.DefaultClient}).request(t.Context(), http.MethodGet, "/", nil, http.StatusOK); err == nil {
+		t.Fatal("request accepted an invalid URL")
+	}
+}
+
+func TestConnectorDrainReaderAndLiveRuntimeDelegations(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	mock.ExpectQuery("SELECT id, name, note, counter").WillReturnRows(
+		sqlmock.NewRows([]string{"id", "name", "note", "counter"}).AddRow(int64(1), "one", nil, int64(1)),
+	)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = w.Write([]byte(`{"offsets":[{"partition":{"kafka_topic":"mo_issue_29848_events","kafka_partition":0},"offset":{"kafka_offset":1}}]}`))
+	}))
+	defer server.Close()
+	client, err := kgo.NewClient(kgo.SeedBrokers("127.0.0.1:1"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+	reader := connectorDrainReader{
+		db: db, admin: kadm.NewClient(client),
+		connect: &connectClient{base: server.URL, client: server.Client()}, name: "fixture",
+	}
+	if rows, err := reader.rows(t.Context()); err != nil || len(rows) != 1 {
+		t.Fatalf("connectorDrainReader.rows = %+v, %v", rows, err)
+	}
+	if offset, err := reader.restOffset(t.Context()); err != nil || offset != 1 {
+		t.Fatalf("connectorDrainReader.restOffset = %d, %v", offset, err)
+	}
+	canceled, cancel := context.WithCancel(t.Context())
+	cancel()
+	if _, _, err := reader.groupOffset(canceled); err == nil {
+		t.Fatal("connectorDrainReader.groupOffset unexpectedly succeeded")
+	}
+	if _, _, err := reader.logEndOffset(canceled); err == nil {
+		t.Fatal("connectorDrainReader.logEndOffset unexpectedly succeeded")
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatal(err)
+	}
+
+	for _, statement := range []string{
+		"CREATE DATABASE IF NOT EXISTS " + databaseName,
+		"DROP TABLE IF EXISTS " + databaseName + "." + tableName,
+		"CREATE TABLE " + databaseName + "." + tableName,
+	} {
+		mock.ExpectExec(statement).WillReturnResult(sqlmock.NewResult(0, 1))
+	}
+	runtime := liveRecoveryRuntime{connectorDrainReader: reader, kafka: client, opts: runOptions{composeProject: "fixture", composeBase: t.TempDir() + "/base.yaml", composeConnect: t.TempDir() + "/connect.yaml"}}
+	if err := runtime.prepareTarget(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	if err := runtime.createTopic(ctx); err == nil {
+		t.Fatal("live runtime createTopic unexpectedly succeeded")
+	}
+	if _, err := runtime.produceRows(ctx, []row{{ID: 1, Name: "one", Counter: 1}}); err == nil {
+		t.Fatal("live runtime produceRows unexpectedly succeeded")
+	}
+	if _, err := runtime.produceTombstone(ctx, 1); err == nil {
+		t.Fatal("live runtime produceTombstone unexpectedly succeeded")
+	}
+	dockerDir := t.TempDir()
+	docker := filepath.Join(dockerDir, "docker")
+	if err := os.WriteFile(docker, []byte("#!/bin/sh\ncase \"$1\" in\ncompose) printf 'container-1\\n' ;;\nkill) ;;\ninspect) printf 'exited\\n' ;;\nesac\n"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("PATH", dockerDir+string(os.PathListSeparator)+os.Getenv("PATH"))
+	if err := runtime.killWorker(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	if err := runtime.startWorker(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestDecodeAndKafkaHelperFailures(t *testing.T) {
+	if _, err := decodeKeyID([]byte("{")); err == nil {
+		t.Fatal("decodeKeyID accepted malformed JSON")
+	}
+	if got, err := decodeKeyID([]byte(`{"payload":{"id":17}}`)); err != nil || got != 17 {
+		t.Fatalf("decodeKeyID = %d, %v, want 17", got, err)
+	}
+	client, err := kgo.NewClient(kgo.SeedBrokers("127.0.0.1:1"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	records := []*kgo.Record{{Topic: topicName, Key: []byte(`{"payload":{"id":1}}`), Value: []byte(`{}`)}}
+	if _, err := produce(ctx, client, records); err == nil {
+		t.Fatal("produce unexpectedly succeeded with a canceled context")
+	}
+	if _, err := produceRows(ctx, client, []row{{ID: 1, Name: "one", Counter: 1}}); err == nil {
+		t.Fatal("produceRows unexpectedly succeeded with a canceled context")
+	}
+	if _, err := produceTombstone(ctx, client, 1); err == nil {
+		t.Fatal("produceTombstone unexpectedly succeeded with a canceled context")
+	}
+	_, admin := canceledKafkaAdmin(t)
+	if err := createTopic(ctx, admin); err == nil {
+		t.Fatal("createTopic unexpectedly succeeded with a canceled context")
+	}
+	if _, _, err := groupOffset(ctx, admin, "fixture"); err == nil {
+		t.Fatal("groupOffset unexpectedly succeeded with a canceled context")
+	}
+	if _, _, err := topicLogEndOffset(ctx, admin); err == nil {
+		t.Fatal("topicLogEndOffset unexpectedly succeeded with a canceled context")
+	}
+}
+
+func TestKafkaHelpersAgainstInProcessBroker(t *testing.T) {
+	cluster, err := kfake.NewCluster(kfake.NumBrokers(1))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cluster.Close()
+	client, err := kgo.NewClient(kgo.SeedBrokers(cluster.ListenAddrs()...), kgo.RequiredAcks(kgo.AllISRAcks()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+	admin := kadm.NewClient(client)
+	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+	defer cancel()
+	if err := createTopic(ctx, admin); err != nil {
+		t.Fatal(err)
+	}
+	if err := createTopic(ctx, admin); err != nil {
+		t.Fatalf("idempotent topic creation failed: %v", err)
+	}
+	records, err := produceRows(ctx, client, []row{{ID: 1, Name: "one", Counter: 1}, {ID: 2, Name: "two", Note: stringPointer(""), Counter: 2}})
+	if err != nil || len(records) != 2 || records[0].Offset != 0 || records[1].Offset != 1 {
+		t.Fatalf("produceRows records = %+v, err=%v", records, err)
+	}
+	tombstone, err := produceTombstone(ctx, client, 1)
+	if err != nil || len(tombstone) != 1 || !tombstone[0].Tombstone || tombstone[0].Offset != 2 {
+		t.Fatalf("produceTombstone records = %+v, err=%v", tombstone, err)
+	}
+	end, found, err := topicLogEndOffset(ctx, admin)
+	if err != nil || !found || end != 3 {
+		t.Fatalf("topicLogEndOffset = %d, found=%v, err=%v", end, found, err)
+	}
+	responses, err := admin.CommitOffsets(ctx, "fixture", (kadm.OffsetsList{{Topic: topicName, Partition: 0, At: 3}}).Offsets())
+	if err != nil || !responses.Ok() {
+		t.Fatalf("commit fixture offset = %+v, err=%v", responses, err)
+	}
+	group, found, err := groupOffset(ctx, admin, "fixture")
+	if err != nil || !found || group != 3 {
+		t.Fatalf("groupOffset = %d, found=%v, err=%v", group, found, err)
+	}
+}
+
+func TestWaitForDatabaseFailureAndPortParsing(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	if err := waitForDatabase(ctx, alwaysFailPinger{}); err == nil || !strings.Contains(err.Error(), "last ping") {
+		t.Fatalf("waitForDatabase error = %v, want last ping context", err)
+	}
+	for _, test := range []struct {
+		dsn  string
+		want string
+	}{
+		{dsn: "root@tcp(127.0.0.1:6601)/", want: "6601"},
+		{dsn: "root@tcp(127.0.0.1)/", want: "6001"},
+		{dsn: "root@/", want: "6001"},
+	} {
+		if got := portFromDSN(test.dsn); got != test.want {
+			t.Fatalf("portFromDSN(%q) = %q, want %q", test.dsn, got, test.want)
+		}
+	}
+}
+
+func TestPollAndWriteReportFailureBranches(t *testing.T) {
+	if err := poll(t.Context(), "scripted check", time.Second, func(context.Context) (bool, error) {
+		return false, errors.New("check failed")
+	}); err == nil || !strings.Contains(err.Error(), "scripted check") {
+		t.Fatalf("poll error = %v, want scripted check failure", err)
+	}
+	reportDir := t.TempDir()
+	if err := os.Mkdir(filepath.Join(reportDir, "report.json"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := writeReport(reportDir, runReport{Status: "failed"}); err == nil {
+		t.Fatal("writeReport replaced an existing report directory")
+	}
+}
+
+type alwaysFailPinger struct{}
+
+func (alwaysFailPinger) PingContext(context.Context) error { return errors.New("ping failed") }
+
+func canceledKafkaAdmin(t *testing.T) (*kgo.Client, *kadm.Client) {
+	t.Helper()
+	client, err := kgo.NewClient(kgo.SeedBrokers("127.0.0.1:1"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(client.Close)
+	return client, kadm.NewClient(client)
+}
+
+func TestOwnedConnectDockerHelpers(t *testing.T) {
+	composeDir := t.TempDir()
+	opts := runOptions{composeProject: "fixture", composeBase: filepath.Join(composeDir, "base.yaml"), composeConnect: filepath.Join(composeDir, "connect.yaml")}
+	dockerDir := t.TempDir()
+	docker := filepath.Join(dockerDir, "docker")
+	if err := os.WriteFile(docker, []byte("#!/bin/sh\ncase \"$1\" in\ncompose) case \"$*\" in *\"ps -q connect\"*) printf 'container-1\\n' ;; *\"start connect\"*) printf 'started\\n' ;; *) printf 'compose-output\\n' ;; esac ;;\nkill) ;;\ninspect) printf 'exited\\n' ;;\nesac\n"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("PATH", dockerDir+string(os.PathListSeparator)+os.Getenv("PATH"))
+	if output, err := composeOutput(t.Context(), opts, "config"); err != nil || !strings.Contains(output, "compose-output") {
+		t.Fatalf("composeOutput = %q, %v", output, err)
+	}
+	if err := startOwnedConnect(t.Context(), opts); err != nil {
+		t.Fatal(err)
+	}
+	if err := killOwnedConnect(t.Context(), opts); err != nil {
+		t.Fatal(err)
+	}
+
+	emptyDockerDir := t.TempDir()
+	emptyDocker := filepath.Join(emptyDockerDir, "docker")
+	if err := os.WriteFile(emptyDocker, []byte("#!/bin/sh\nif [ \"$1\" = compose ]; then exit 0; fi\n"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("PATH", emptyDockerDir+string(os.PathListSeparator)+os.Getenv("PATH"))
+	if err := killOwnedConnect(t.Context(), opts); err == nil || !strings.Contains(err.Error(), "container was not found") {
+		t.Fatal("killOwnedConnect accepted an empty container id")
+	}
+
+	failingDockerDir := t.TempDir()
+	failingDocker := filepath.Join(failingDockerDir, "docker")
+	if err := os.WriteFile(failingDocker, []byte("#!/bin/sh\nprintf 'docker failed\\n'\nexit 17\n"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("PATH", failingDockerDir+string(os.PathListSeparator)+os.Getenv("PATH"))
+	if _, err := composeOutput(t.Context(), opts, "config"); err == nil {
+		t.Fatal("composeOutput accepted a failed docker command")
+	}
+}
+
+func TestComposeOutputResolvesRelativePathsBeforeChangingDirectory(t *testing.T) {
+	_, testFile, _, ok := runtime.Caller(0)
+	if !ok {
+		t.Fatal("could not locate test source")
+	}
+	repoRoot := filepath.Clean(filepath.Join(filepath.Dir(testFile), "..", ".."))
+	originalWorkingDirectory, err := os.Getwd()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Chdir(repoRoot); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if err := os.Chdir(originalWorkingDirectory); err != nil {
+			t.Errorf("restore working directory: %v", err)
+		}
+	})
+
+	const relativeBase = "etc/launch-kafka-local/compose.yaml"
+	const relativeConnect = "etc/launch-kafka-local/compose.connect.yaml"
+	basePath, err := filepath.Abs(relativeBase)
+	if err != nil {
+		t.Fatal(err)
+	}
+	connectPath, err := filepath.Abs(relativeConnect)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, path := range []string{basePath, connectPath} {
+		if !filepath.IsAbs(path) {
+			t.Fatalf("Compose path %q is not absolute", path)
+		}
+		if _, err := os.Stat(path); err != nil {
+			t.Fatalf("expected Compose file %q to exist: %v", path, err)
+		}
+	}
+
+	dockerDir := t.TempDir()
+	logPath := filepath.Join(t.TempDir(), "docker-args.txt")
+	dockerPath := filepath.Join(dockerDir, "docker")
+	dockerScript := "#!/bin/sh\nprintf '%s\\n' \"$PWD\" >\"$TEST_DOCKER_LOG\"\nprintf '%s\\n' \"$@\" >>\"$TEST_DOCKER_LOG\"\n"
+	if err := os.WriteFile(dockerPath, []byte(dockerScript), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("PATH", dockerDir+string(os.PathListSeparator)+os.Getenv("PATH"))
+	t.Setenv("TEST_DOCKER_LOG", logPath)
+	opts := runOptions{
+		composeProject: "fixture",
+		composeBase:    relativeBase,
+		composeConnect: relativeConnect,
+	}
+	if output, err := composeOutput(t.Context(), opts, "config"); err != nil {
+		t.Fatalf("composeOutput = %q, %v", output, err)
+	}
+	log, err := os.ReadFile(logPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	lines := strings.Split(strings.TrimSpace(string(log)), "\n")
+	want := []string{filepath.Dir(basePath), "compose", "-p", "fixture", "-f", basePath, "-f", connectPath, "config"}
+	if !reflect.DeepEqual(lines, want) {
+		t.Fatalf("docker working directory and arguments = %q, want %q", lines, want)
+	}
+}
+
+func TestRunRejectsUnsafeOrUnavailableSetup(t *testing.T) {
+	report := &runReport{Phases: []phase{}}
+	if err := run(t.Context(), runOptions{reportDir: t.TempDir()}, report); err == nil || !strings.Contains(err.Error(), "compose project is required") {
+		t.Fatal("run accepted an empty compose project")
+	}
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	report = &runReport{Phases: []phase{}}
+	opts := runOptions{
+		composeProject: "fixture",
+		dsn:            "root:111@tcp(127.0.0.1:1)/?timeout=1ms",
+		reportDir:      t.TempDir(),
+		phaseTimeout:   time.Second,
+	}
+	if err := run(ctx, opts, report); err == nil || !strings.Contains(err.Error(), "MatrixOne SQL readiness") {
+		t.Fatalf("run setup error = %v, want MatrixOne readiness failure", err)
+	}
+}
+
+func TestRunMainValidatesArgumentsAndRecordsFailures(t *testing.T) {
+	if code := runMain([]string{"-h"}); code != 0 {
+		t.Fatalf("runMain help code = %d, want 0", code)
+	}
+	if code := runMain([]string{"-timeout=0"}); code != 2 {
+		t.Fatalf("runMain timeout validation code = %d, want 2", code)
+	}
+	if code := runMain([]string{"-unknown-flag"}); code != 2 {
+		t.Fatalf("runMain flag parsing code = %d, want 2", code)
+	}
+	badReportPath := filepath.Join(t.TempDir(), "report-file")
+	if err := os.WriteFile(badReportPath, []byte("occupied"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if code := runMain([]string{"-report-dir", badReportPath}); code != 1 {
+		t.Fatalf("runMain report setup code = %d, want 1", code)
+	}
+	if code := runMain([]string{"-report-dir", t.TempDir()}); code != 1 {
+		t.Fatalf("runMain setup failure code = %d, want 1", code)
+	}
+}
+
+func TestRunUsesInjectableReadinessAndRecoveryDependencies(t *testing.T) {
+	db, mock, err := sqlmock.New(sqlmock.MonitorPingsOption(true))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	mock.ExpectPing().WillReturnError(nil)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		switch request.URL.Path {
+		case "/":
+			_, _ = w.Write([]byte(`{"version":"3.8.1","kafka_cluster_id":"cluster-1"}`))
+		case "/connector-plugins":
+			_, _ = w.Write([]byte(`[{"class":"io.debezium.connector.jdbc.JdbcSinkConnector","type":"sink","version":"2.6"}]`))
+		default:
+			http.NotFound(w, request)
+		}
+	}))
+	defer server.Close()
+	client, err := kgo.NewClient(kgo.SeedBrokers("127.0.0.1:1"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+	recoveryCalled := false
+	dependencies := runDependencies{
+		openDatabase: func(string) (*sql.DB, error) { return db, nil },
+		newKafkaClient: func(...kgo.Opt) (*kgo.Client, error) {
+			return client, nil
+		},
+		runRecoveryFunc: func(_ context.Context, opts runOptions, report *runReport, connect connectClient, runtime recoveryRuntime) error {
+			recoveryCalled = true
+			if opts.connectorName != "fixture" || report.ConnectVersion != "3.8.1" || connect.base != server.URL || runtime == nil {
+				return errors.New("run did not pass initialized dependencies")
+			}
+			return nil
+		},
+	}
+	report := &runReport{Phases: []phase{}}
+	opts := runOptions{
+		composeProject: "fixture", connectorName: "fixture", connectURL: server.URL,
+		reportDir: t.TempDir(), dsn: "ignored", phaseTimeout: time.Second,
+	}
+	if err := run(t.Context(), opts, report, dependencies); err != nil {
+		t.Fatal(err)
+	}
+	if !recoveryCalled || report.ConnectVersion != "3.8.1" || report.KafkaClusterID != "cluster-1" || report.JDBCPluginVersion != "2.6" {
+		t.Fatalf("run report/dependency state = %+v, recovery called=%v", report, recoveryCalled)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestRunReportsDependencyFailureBranches(t *testing.T) {
+	baseOptions := func(reportDir string, connectURL string) runOptions {
+		return runOptions{composeProject: "fixture", connectURL: connectURL, reportDir: reportDir, phaseTimeout: 20 * time.Millisecond}
+	}
+	newDatabase := func(t *testing.T) *sql.DB {
+		t.Helper()
+		db, mock, err := sqlmock.New(sqlmock.MonitorPingsOption(true))
+		if err != nil {
+			t.Fatal(err)
+		}
+		mock.ExpectPing().WillReturnError(nil)
+		t.Cleanup(func() {
+			if err := mock.ExpectationsWereMet(); err != nil {
+				t.Error(err)
+			}
+			_ = db.Close()
+		})
+		return db
+	}
+	root := `{"version":"3.8.1","kafka_cluster_id":"cluster-1"}`
+	plugins := `[{"class":"io.debezium.connector.jdbc.JdbcSinkConnector","type":"sink","version":"2.6"}]`
+	newServer := func(t *testing.T, rootBody, pluginsBody string, pluginStatus int) *httptest.Server {
+		t.Helper()
+		return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
+			w.Header().Set("Content-Type", "application/json")
+			switch request.URL.Path {
+			case "/":
+				w.WriteHeader(http.StatusOK)
+				_, _ = w.Write([]byte(rootBody))
+			case "/connector-plugins":
+				w.WriteHeader(pluginStatus)
+				_, _ = w.Write([]byte(pluginsBody))
+			default:
+				http.NotFound(w, request)
+			}
+		}))
+	}
+
+	t.Run("open database", func(t *testing.T) {
+		dependencies := defaultRunDependencies()
+		dependencies.openDatabase = func(string) (*sql.DB, error) { return nil, errors.New("driver unavailable") }
+		report := &runReport{Phases: []phase{}}
+		err := run(t.Context(), baseOptions(t.TempDir(), "http://127.0.0.1:1"), report, dependencies)
+		if err == nil || !strings.Contains(err.Error(), "open MatrixOne DSN") {
+			t.Fatalf("run error = %v, want database open failure", err)
+		}
+	})
+
+	t.Run("initial report stage", func(t *testing.T) {
+		db, _, err := sqlmock.New()
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer db.Close()
+		badPath := filepath.Join(t.TempDir(), "report-file")
+		if err := os.WriteFile(badPath, []byte("occupied"), 0o644); err != nil {
+			t.Fatal(err)
+		}
+		dependencies := defaultRunDependencies()
+		dependencies.openDatabase = func(string) (*sql.DB, error) { return db, nil }
+		report := &runReport{Phases: []phase{}}
+		runErr := run(t.Context(), baseOptions(badPath, "http://127.0.0.1:1"), report, dependencies)
+		if runErr == nil {
+			t.Fatal("run accepted a report path that is a file")
+		}
+	})
+
+	t.Run("connect readiness", func(t *testing.T) {
+		db := newDatabase(t)
+		server := newServer(t, `{"version":""}`, plugins, http.StatusOK)
+		defer server.Close()
+		dependencies := defaultRunDependencies()
+		dependencies.openDatabase = func(string) (*sql.DB, error) { return db, nil }
+		report := &runReport{Phases: []phase{}}
+		err := run(t.Context(), baseOptions(t.TempDir(), server.URL), report, dependencies)
+		if err == nil || !strings.Contains(err.Error(), "wait for Kafka Connect") {
+			t.Fatalf("run error = %v, want Connect readiness failure", err)
+		}
+	})
+
+	t.Run("plugin REST failure", func(t *testing.T) {
+		db := newDatabase(t)
+		server := newServer(t, root, "plugin unavailable", http.StatusServiceUnavailable)
+		defer server.Close()
+		dependencies := defaultRunDependencies()
+		dependencies.openDatabase = func(string) (*sql.DB, error) { return db, nil }
+		report := &runReport{Phases: []phase{}}
+		err := run(t.Context(), baseOptions(t.TempDir(), server.URL), report, dependencies)
+		if err == nil || !strings.Contains(err.Error(), "list Kafka Connect plugins") {
+			t.Fatalf("run error = %v, want plugin REST failure", err)
+		}
+	})
+
+	t.Run("missing sink plugin", func(t *testing.T) {
+		db := newDatabase(t)
+		server := newServer(t, root, `[]`, http.StatusOK)
+		defer server.Close()
+		dependencies := defaultRunDependencies()
+		dependencies.openDatabase = func(string) (*sql.DB, error) { return db, nil }
+		report := &runReport{Phases: []phase{}}
+		err := run(t.Context(), baseOptions(t.TempDir(), server.URL), report, dependencies)
+		if err == nil || !strings.Contains(err.Error(), "required sink plugin") {
+			t.Fatalf("run error = %v, want missing plugin failure", err)
+		}
+	})
+
+	t.Run("Kafka client creation", func(t *testing.T) {
+		db := newDatabase(t)
+		server := newServer(t, root, plugins, http.StatusOK)
+		defer server.Close()
+		dependencies := defaultRunDependencies()
+		dependencies.openDatabase = func(string) (*sql.DB, error) { return db, nil }
+		dependencies.newKafkaClient = func(...kgo.Opt) (*kgo.Client, error) { return nil, errors.New("broker unavailable") }
+		report := &runReport{Phases: []phase{}}
+		err := run(t.Context(), baseOptions(t.TempDir(), server.URL), report, dependencies)
+		if err == nil || !strings.Contains(err.Error(), "create Kafka client") {
+			t.Fatalf("run error = %v, want Kafka client failure", err)
+		}
+	})
+}